Quantopian 入门系列二 - 流水线 (上)

本文含 8045 字，18 图表截屏

建议阅读 42 分钟

引言

在上贴〖Quantopian 系列一〗我们初探了的流水线（pipeline），本帖我们就把它揉碎了讲。

很多交易算法都需要重复的做以下几个步骤

1. 对于已知集合中的每个资产，计算它们在不同移动窗口下的 N 个统计量
2. 根据『1』中计算的值选择可交易的子资产集（subset）
3. 根据『2』中选择的资产集上计算所需的投资组合权重
4. 根据『3』中计算的权重下订单

能高效地执行上述操作存在几个技术难点，包括：

• 要高效的查询大量资产
• 要在大量资产上进行计算
• 要能处理调整（拆分和股息）和资产退市

流水线就是来解决这些难点的。

在上贴中，我们了解到 Quantopian 有研究环境（research）和回测环境（backtest），我们可以在前者中快速迭代不同的交易策略，然后再后者构建下订单对其策略进行回测。

两个环境都需要流水线，好消息是，在两个环境中，构造流水线是相同的，唯一不同的是其运行方式

• 研究环境：需要设定起始日和终止日来运行流水线
• 回测环境：不需要设定起始日和终止日来运行流水线，因此在每个回测日都要跑一次流水线

目录如下：

1. 简介
2. 因子
3. 筛选器
4. 分类器
5. 蒙面法
6. 数据集
7. 自定义因子
8. 回测

由于内容较多，本帖分上下两贴，上贴讲 1 至 4 节，下帖讲 5 至 8 节。

1

简介

在流水线中，我们可以在同时在多个资产中的多维特征上定义一系列运算，而这些计算可分为三大类：

• 因子（factor）
• 筛选器（filter）
• 分类器（classifer）

三者的相同点：都是从资产和时点产生值的函数

三者的不同点：产生值的类型不同

因子

因子是从「资产和时点」到数值的函数。

关于因子的两个简单示例：

1. 资产的最新价格，给定资产和特定时点，得到的最新价格是个数值。
2. 资产的 10 天平均交易量，给定资产和特定时点（10 个时点），得到的平均交易量是个数值。

因子最常用于以下几种方式：

• 计算目标权重
• 产生交易信号
• 构造更复杂的因子
• 构造筛选器

筛选器

筛选器是从「资产和时点」到布尔值的函数。

关于筛选器的一个示例：判断资产价格是否低于 10 美元，在给定资产和时点的情况下，此结果只能为真（True）或假（False），是个布尔值。

筛选器最常用于包括或剔除的某些资产集。

分类器

分类器是从「资产和时点」到分类值的函数。

具体来讲，分类器产生的分类值可以是字符串（string）或整数（integer）。注意这里整数是指整数标签，例如行业代码（sector code）。

关于分类器的一个示例：正在交易的资产所在交易所的代号。

分类器最常用于对资产进行分组。

数据集

流水线可以在多种数据上进行计算，比如 OHLC 数据、交易量数据、基本面数据和情绪数据等。我们将在后面会介绍每类数据集。

接下来第 2, 4, 6, 10 节讨论因子，第 3, 5, 7 节讨论筛选器，第 8 节讨论分类器，第 9 节讨论数据集，第 11 节结合所有内容构建一个完整的策略做回测。

2

因子

因子本质上是一个函数，将资产和时点两个自变量转化成一个数值型变量（numeric variable）。抽象形式如下：

F(asset, timestamp) -> float

其中 asset 是一列数据，timestamp 是一个窗口长度。

Quantopian 里面有一组内置因子（built-in factors）。比如在计算最近 10 天的平均收盘价时，我们可以使用 SimpleMovingAverage 内置因子来计算指定窗口长度（10天）内输入数据（收盘价）的平均值。为此，我们需要导入内置的 SimpleMovingAverage 因子和 USEquityPricing 数据集。

首先引用它们，加上定义流水线的 Pipeline 和运行流水线的 run_pipeline。

创建因子

当创建「移动均值」的因子时，我们只用调用 SimpleMovingAverage 构造函数，来实例化该因子。构造函数需要设定两个参数：

1. input - 一组数据对象的列表
2. window_length - 一个整数，表示移动平均值计算应用多少天的数据

下行代码创建了用于计算「10 天美股平均收盘价」的因子。

mean_close_10 = SimpleMovingAverage(inputs=[USEquityPricing.close],                       window_length=10)

把因子加入流水线

定义好因子后，只需在传给 Pipeline 里面的 columns 参数，代码如下：

看看流水线在 2019-11-25 上运行的结果（需要将起始日和终止日都设成 2019-11-25），打印其首尾五行：

result = run_pipeline(make_pipeline(), '2019-11-25', '2019-11-25')result.head().append(result.tail())

流水线的产出就是个多层的数据帧，第 0 层的行标签是时间，第 1 层的行标签是资产代号，列标签就是上面 Pipeline 里面赋值给 columns 参数的字典的键。

上面结果只有一天，因此第 0 层只显示了 2019-11-25，接下来我们看看流水线在 2019-11-25 到 2019-11-27 上运行的结果，并打印其首尾五行：

result = run_pipeline(make_pipeline(), '2019-11-25', '2019-11-27')result.head().append(result.tail())

除了在 Pipeline 里面用 columns 参数加因子，我们还可以用 Pipeline.add 方法，语法如下：

>>> my_pipe = Pipeline()

>>> f1 = SomeFactor(...)

>>> my_pipe.add(f1, 'f1')

Latest

Quanopian 里最常见的内置因子是 latest，就是获取数据序列中最新的值。由于该因子用的太频繁了，因此我们不是构建函数来实例化它，而是用 .latest 方法来实例化。代码如下：

看看结果：

result = run_pipeline(make_pipeline(), '2019-11-25', '2019-11-25')result.head().append(result.tail())

这是返回的数据帧里面有两个列标签，一个是 10 天平均值，一个是当日最新值。

需要注意的是，.latest 方法不仅仅只用到因子上，还可以用到其它方面，接下来会介绍。

默认输入

对于有默认输入（default input）的因子，我们可以不指定输入。

比如在计算以交易量加权平均价格（VWAP）因子时，我们需要收盘价和交易量，分别从USEquityPricing.close 和 USEquityPricing.volume 获取，它们都是默认因子，因此我们在计算 VWAP 时可以不指定它们。代码如下：

from quantopian.pipeline.factors import VWAPvwap = VWAP(window_length=10)

回顾一下并对比 SimpleMovingAverage 的代码，

mean_close_10 = SimpleMovingAverage(inputs=[USEquityPricing.close],                       window_length=10)

在调用 VWAP 时，没有设置 inputs。

组合因子

多个因子可以组合成新的因子，通过任何内置的数学运算符（+, -, * 等）。例如，构造一个可以计算其他两个因子的平均值的因子很简单：

    >>> f1 = SomeFactor(...)
    >>> f2 = SomeOtherFactor(...)
    >>> average = (f1 + f2) / 2.0

接下来，我们来创建一个 relative_difference 的因子，来计算 10 天平均和30 天平均的相对差。

首先定义两者，注意 window_length 设置为 10 和 30。

mean_close_10 = SimpleMovingAverage(inputs=[USEquityPricing.close],                        window_length=10)mean_close_30 = SimpleMovingAverage(inputs=[USEquityPricing.close],                       window_length=30)

在计算出两者的相对差。

percent_difference = (mean_close_10 - mean_close_30) / mean_close_30

显然 percent_difference 还是一个因子，一个组合因子。只要是因子，都可以写到流水线中，代码如下：

打印结果。

result = run_pipeline(make_pipeline(), '2019-11-25', '2019-11-25')result.head().append(result.tail())

这时输出的数据帧的列标签就是 percent_difference 了。

3

筛选器

筛选器（filter）本质上是一个函数，将资产和时点两个自变量转化成一个布尔型变量（boolean variable）的函数。抽象形式如下：

F(asset, timestamp) -> boolean

在流水线中，筛选器用于缩小资产范围。有两种常见的创建筛选器的方法：

1. 比较运算符方法
2. 因子对象中的方法

比较运算符

例一：创建一个筛选器 close_price_filter，当最新收盘价高于 $20 时，返回 True。

last_close_price = USEquityPricing.close.latestclose_price_filter = last_close_price > 20

例二：创建一个筛选器 mean_crossover_filter，当 10 天均值低于30 天均值时，返回 True。

mean_close_10 = SimpleMovingAverage(inputs=[USEquityPricing.close],                      window_length=10)mean_close_30 = SimpleMovingAverage(inputs=[USEquityPricing.close],                      window_length=30)mean_crossover_filter = mean_close_10 < mean_close_30

在使用筛选器时，每个资产在每个时点都会得到自己对应的 True 和 False。

因子对象中的方法

以 top 函数举例，factor.top(n) 方法产生一个筛选器，它针对给定因子的前n 个资产返回 True。下例的筛选器每天针对正好 200 种资产返回 True，表示这些资产的最新收盘价在所有已知资产的最新收盘价中排名前 200。

last_close_price = USEquityPricing.close.latesttop_close_price_filter = last_close_price.top(200)

具体筛选实例

在下例中，我们想筛选出「过去 30 天平均交易额大于 $10,000,000」的资产，首先引入 AverageDollarVolume

from quantopian.pipeline.factors import AverageDollarVolume

再实例化「平均交易额」这个因子，注意AverageDollarVolume 用 USEquityPricing.close 和 USEquityPricing.volume 作为默认参数，因此不用特意设定（还记得第 2 章讲的默认输入吗？）。

dollar_volume = AverageDollarVolume(window_length=30)

再创建布尔变量类型的筛选条件：

high_dollar_volume = (dollar_volume > 10000000)

全部写到流水线中的代码如下：

看看结果：

result = run_pipeline(make_pipeline(), '2019-11-25', '2019-11-25')print('Number of securities: %d' % len(result))result.head().append(result.tail())

返回的数据帧多了 high_dollar_volume 的列标签，注意到我们打印出所有资产的数目，8880 个。

默认情况下，流水线每天会为 Quantopian 数据库中的每个资产生成各种计算值。但很多时候，我们只关心满足特定条件的一部分资产（比如我们只关心日交易量大过某个阈值的股票）。这是可以通过 screen 关键字将筛选器传递给流水线，相当于告诉流水线忽略掉筛选器产生 False 的资产。

要筛选管道输出的 30 天平均金额大于 $10,000,000 的股票，我们可以把 high_dollar_volume 过滤器传给 screen 参数，见以下代码并注意高亮部分。

看看结果：

result = run_pipeline(make_pipeline(), '2019-11-25', '2019-11-25')print('Number of securities that passed the filter: %d' % len(result))result.head().append(result.tail())

注意筛选过后资产的数目只有 2041 个了，远少于没筛选过的 8880 个。

反转筛选器

符号 ~ 用于反转帅选器，即把原来 True 变成 False 和原来的 False 变成 True。比如

low_dollar_volume = ~high_dollar_volume

这个筛选器 low_dollar_filter 在「过去 30 天的平均交易额小于等于 $1,000,000」的时候返回 True。

组合筛选器

和因子相同的是，筛选器也可以组合起来用；和因子不同的是，我们使用 ＆, | 运算符而不是 +, -, * 运算符。

比如说我们想筛选出「过去 30 天的平均交易额」排前 10% 和「最新收盘价」高于 $20 的资产，我们首先分别定义出这两个筛选器

筛选器一：过去 30 天的平均交易额排 10%，用 percentile_between 方法。

dollar_volume = AverageDollarVolume(window_length=30)high_dollar_volume = dollar_volume.percentile_between(90, 100)

筛选器二：「最新收盘价」高于 $20，用 AverageDollarVolume 因子。

latest_close = USEquityPricing.close.latestabove_20 = latest_close > 20

将筛选器一 high_dollar_volume 和筛选器二 above_20 合并以来，用 & 操作符，组合成新的筛选器 tradeable_filter。

tradeable_filter = high_dollar_volume & above_20

把 tradeable_filter过滤器传给 screen 参数，见以下代码并注意高亮部分。

看看结果：

result = run_pipeline(make_pipeline(), '2019-11-25', '2019-11-25')print('Number of securities that passed the filter: %d' % len(result))result.head().append(result.tail())

用组合筛选器将资产范围进一步缩小到 787 个。

4

分类器

分类器（classifier）本质上是一个函数，将资产和时点两个自变量转化成一个分类型变量（categorical variable）的函数。抽象形式如下：

F(asset, timestamp) -> category

类别 category 可以是一个字符串标签，也可以是一个整数标签。

分类器常见的例子包括交易所的 ID 和行业代号（sector code），如下：

from quantopian.pipeline.data import Fundamentalsexchange = Fundamentals.exchange_id.latestexchange

Latest([Fundamentals<US>.exchange_id], 1)

from quantopian.pipeline.classifiers.fundamentals import Sector  morningstar_sector = Sector()morningstar_sector 

Sector([Fundamentals<US>.morningstar_sector_code], 1)

很明显，这两个分类型变量都是整数标签。

用分类器构建筛选器

这个逻辑很好懂，分类器包含很多类别，比如类 1、类 2、...、类 n。筛选器就是一个条件，可以选择分类器的若干类别来做筛选。

举个例子，如果我们想筛选出在纽约股票交易所（NYSE）上交易的证券，我们可以用分类器 exchange 中的 eq() 方法，代码如下：

nyse_filter = exchange.eq('NYS')

除了 eq() 方法，还有几个常见的用分类器构建筛选器的方法：

• 筛选器 = 分类器.isnull()
• 筛选器 = 分类器.notnull()
• 筛选器 = 分类器.startwith(字符串)
• 筛选器 = 分类器.endwith(字符串)

用因子构建分类器

分类器也可以由因子构建，其中最常见的是分位数方法（quantiles）。该方法把箱数（bin counts）记做 n，并为因子输出中的每个非 NaN 数据点分配从 0 到 n-1 的标签，并返回带有这些标签的分类器（NaN数据标记为 -1）。

首先了解分位数中一些中英文名词和函数的映射如下：

• 四分位数，quartiles (quantiles(4))
• 五分位数，quintiles (quantiles(5))
• 十分位数，deciles (quantiles(10))

下面代码第一行用 AverageDollarVolume 因子加上 deciles() 方法构建一个含十类的分类器。第二行用 eq() 方法筛选出第 9 类，即选取 ADV 排前 10% 的资产。

dollar_volume_decile = AverageDollarVolume(window_length=10).deciles()top_decile = (dollar_volume_decile.eq(9))

将两个筛选器（一个交易所的 nyse_filter 和一个 ADV 前 10% 的 top_decile）传给 screen 参数，注意下面代码高亮部分。

看看结果：

result = run_pipeline(make_pipeline(), '2019-11-25', '2019-11-25')print('Number of securities that passed the filter: %d' % len(result))result.head().append(result.tail())

筛选出 491 个资产。

上贴把流水线里因子、筛选器和分类器三大核心讲清楚了，几个重点要记住：

1. 因子（数值）和筛选器（布尔值）可以是单个的，也可以是组合的。
2. 组合因子用的 +, -, *, / 等运算符号，组合筛选器用的是 &, | 等条件符号。
3. 分类器（分类值）可以用来构建筛选器，用 eq(), isnull(), startwith() 等方法。
4. 因子可以用来构建分类器，但最终分类器还是用来构建筛选器。
5. 因子是你最终想收集的数据，筛选器是帮助你得到你想要的资产集合。

一个高度抽象构建流水线的伪代码长成下面这个样子：

def make_pipeline():

计算一堆因子

列出一堆筛选器

生成一堆分类器

分类器生成一堆筛选器

return Pipeline(

columns={'因子字符串'：因子}

screen={'筛选器字符串'：筛选器}

)

我疯狂喜欢高度抽象的东西，具体的东西要用的时候再查资料。人脑不是电脑，记不住那么东西！

下帖继续讲流水线。Stay Tuned！