本文含 8045 字，18 图表截屏
建议阅读 42 分钟
引言
在上贴〖Quantopian 系列一〗我们初探了的流水线（pipeline），本帖我们就把它揉碎了讲。
很多交易算法都需要重复的做以下几个步骤
- 对于已知集合中的每个资产，计算它们在不同移动窗口下的 N 个统计量
- 根据『1』中计算的值选择可交易的子资产集（subset）
- 根据『2』中选择的资产集上计算所需的投资组合权重
- 根据『3』中计算的权重下订单
能高效地执行上述操作存在几个技术难点，包括：
- 要高效的查询大量资产
- 要在大量资产上进行计算
- 要能处理调整（拆分和股息）和资产退市
流水线就是来解决这些难点的。
在上贴中，我们了解到 Quantopian 有研究环境（research）和回测环境（backtest），我们可以在前者中快速迭代不同的交易策略，然后再后者构建下订单对其策略进行回测。
两个环境都需要流水线，好消息是，在两个环境中，构造流水线是相同的，唯一不同的是其运行方式
- 研究环境：需要设定起始日和终止日来运行流水线
- 回测环境：不需要设定起始日和终止日来运行流水线，因此在每个回测日都要跑一次流水线
目录如下：
- 简介
- 因子
- 筛选器
- 分类器
- 蒙面法
- 数据集
- 自定义因子
- 回测
由于内容较多，本帖分上下两贴，上贴讲 1 至 4 节，下帖讲 5 至 8 节。
1
简介
在流水线中，我们可以在同时在多个资产中的多维特征上定义一系列运算，而这些计算可分为三大类：
- 因子（factor）
- 筛选器（filter）
- 分类器（classifer）
三者的相同点：都是从资产和时点产生值的函数
三者的不同点：产生值的类型不同
因子
因子是从「资产和时点」到数值的函数。
关于因子的两个简单示例：
- 资产的最新价格，给定资产和特定时点，得到的最新价格是个数值。
- 资产的 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 构造函数，来实例化该因子。构造函数需要设定两个参数：
- input - 一组数据对象的列表
- 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。
组合因子
多个因子可以组合成新的因子，通过任何内置的数学运算符（ , -, * 等）。例如，构造一个可以计算其他两个因子的平均值的因子很简单：
代码语言：javascript复制 >>> 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
在流水线中，筛选器用于缩小资产范围。有两种常见的创建筛选器的方法：
- 比较运算符方法
- 因子对象中的方法
比较运算符
例一：创建一个筛选器 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.latestexchangeLatest([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 个资产。
上贴把流水线里因子、筛选器和分类器三大核心讲清楚了，几个重点要记住：
- 因子（数值）和筛选器（布尔值）可以是单个的，也可以是组合的。
- 组合因子用的 , -, *, / 等运算符号，组合筛选器用的是 &, | 等条件符号。
- 分类器（分类值）可以用来构建筛选器，用
eq(),isnull(),startwith()等方法。 - 因子可以用来构建分类器，但最终分类器还是用来构建筛选器。
- 因子是你最终想收集的数据，筛选器是帮助你得到你想要的资产集合。
一个高度抽象构建流水线的伪代码长成下面这个样子：
def make_pipeline():
计算一堆因子
列出一堆筛选器
生成一堆分类器
分类器生成一堆筛选器
return Pipeline(
columns={'因子字符串'：因子}
screen={'筛选器字符串'：筛选器}
)
我疯狂喜欢高度抽象的东西，具体的东西要用的时候再查资料。人脑不是电脑，记不住那么东西！
下帖继续讲流水线。Stay Tuned！
