????????之前的文章“在Elasticsearch 中回测 RSI 交叉策略”，介紹在Elasticsearch中如何回測 相对强弱指数(RSI)指标交叉交易策略。在本文中，我们将实施随机（Stochastic）指标并衡量其性能，最后与相对强弱指数指标进行比较。 儘管这指標是由 George C. Lane 在 1950 年代後期開發，看似非常古老，但仍然很受歡迎。与 RSI 指标一样，随机指标也是展示在 0 和 100 之间价格的波动动量。所以，它们称为振荡器(Oscillator)。这两个指标都可用于识别超买区域和超卖区域。
????????随机指标将价格变化转化为一种数据，定义为两个距离的比值为 %K (包括最近收盘价C和最近最高价MMax之间的距离，以及最近最高价和最近最低价MMin之间的距离)。随机指标分为两种类型，快速和慢速。假设%Kfast为快速的例子，方程可以改写如下，其中 MMaxn,1 和 MMinn,1 是移动最大值和移动最小值。 对应滑动窗口为n的Elasticsearch移动函数，需要右移1个数据以包含当前数据。滑动窗口 n常用值为14 。
????????与 MACD 类似，随机指标也定义了一条名为 %D 的信号线，它是滑动窗口为3 的%K 简单移动平均值(SMA)。假设%Dfast为快速的例子，方程可以改写如下。
快速比慢速对价格变化更敏感，会产生更多的买入或卖出信号。慢速随机指标%Fslow 及%Dslow可以定义如下：
简单的随机交叉交易策略可以定义为当 %K 线在超买区 (> 80) 越过并且低于%D 线时发出卖出信号，而当 %K 线在超卖区(< 20)中越过并且高于%D 线时发出买入信号。 对于其他值，请耐心等待买入或卖出信号。
?????????使用图表来观察值的变化要容易得多。在本文中，我们将回测应用于Tushare大数据开放社区提供的股票型公募基金，并专注于将 Elasticsearch 作为分析工具。 下面的例子随机选择了"工银研究精选股票" (代码为000803.OF) ，并另外抽取10只股票型基金运行，结果将在最后的段落汇总和展示。数据选自2021年01月01日到2021年05月31日之间的时间范围。在下图中，随机指标 (FK/SK) 和信号线 (FD/SD) 与每日收盘价一起绘制。 在每日价格曲线中，卖出信号的价格标为红色，而买入信号的价格标为蓝色。 如下图所示，慢随机指标产生的信号数量略少于快速随机指标。
????????????在这里，我们展示了一个简单的随机交叉交易策略，并使用 Elasticsearch 来展示实现细节。
- 假设由于成本的限制，只能每一次购买和持有 1 股，并在持有的股份被售出之前不能发生任何交易。
- 当 FK 在超卖区域 (<20) 超过 FD 时买入 1 股。
- 当 FK 在超买区域 (> 80) 低于 FD 时卖出 1 股。
- 在测试期结束时，持有的股票以当前价格兑现。
- 慢速随机交叉交易策略以SK代替FK和以 SD代替FD
????????根据以上的交易策略，快速随机指标有 5 个蓝点和 11 个红点，但只允许进行 3 次买入和 3 次卖出交易。慢速随机指标有 3 个蓝点和 10 个红点，但只允许进行 2 次买入和 2 次卖出交易。让我们描述一下使用 Elasticsearch 的实现。假设有一个填充有数据的 Elasticsearch 索引，其使用的数据映射与上一篇论文中描述的相同。以下步骤演示了 REST API 请求正文的代码。
通过搜索操作收集所有相关文档
使用带有必要条件（must）子句的布林查询（bool query）来收集基金代码为000803.OF，和公告截止日期从2021年01月01日到2021年05月31日的文档。 由于需要计算移动平均值，因此增加了一个月的数据（从2020年12月01日到2020年12月31日）。
{
"query": {
"bool": {
"must": [
{"range": {"end_date": {"gte": "20201201", "lte": "20210531"}}},
{"term": {"ts_code": "000803.OF"}}
]
}
},提取每日的复权单位净值
使用名为Backtest_Stochastic日期直方图（date_histogram）存储桶聚合，并配合参数field（字段）为end_date和interval（间隔）为 1d（1天），提取每日的复权单位净值（adj_nav）。由于子聚合使用管道（pipeline）聚合而无法直接采用文档字段，所以额外使用平均值（avg）聚合获取每日的复权单位净值，聚合名称为Daily。由于买入价格和利润百分比使用单位价格(unit_nav)计算，故此同样建立聚合名称Daily_unit_nav.
"aggs": {
"Backtest_Stochastic": {
"date_histogram": {
"field": "end_date",
"interval": "1d",
"format": "yyyyMMdd"
},
"aggs": {
"Daily": {
"avg": {"field": "adj_nav"}
},
"Daily_unit_nav": {
"avg": {"field": "unit_nav”}
},提取存储桶的日期
由于增加了一个月的数据，而后续操作需要过滤掉额外的文档，因此以存储桶的日期作为筛选限制条件。我们可以使用名为DateStr的最小值聚合间接取得日期，Elasticsearch的日期用新纪元时间(Epoch Time) 表示，并且以毫秒为单位，时区为UTC。
"DateStr": {
"min": {"field": "end_date"}
},选择多于1个文档的bucket
由于没有内插数据，非交易日为了过滤掉空桶），使用名为 SDaily 的“bucket_selector”聚合来选择文档计数大于 0 的桶。
"SDaily": {
"bucket_selector": {
"buckets_path": {"count":"_count"},
"script": "params.count > 0"
}
},计算收盘价的简单移动最大值和最小值
使用两个"moving_fn"聚合，命名为 MMax 和 MMin，参数 window 为 14，参数"buckets_path"为 Daily。 参数"shift"设置为 1 以包括最近的数据。 MMax 和MMin 使用函数MovingFunctions.max() 和MovingFunctions.min() 计算。
"MMax": {
"moving_fn": {
"script": "MovingFunctions.max(values)", "window": 14, "buckets_path": "Daily", "shift":1
}
},
"MMin": {
"moving_fn": {
"script": "MovingFunctions.min(values)", "window": 14, "buckets_path": "Daily", "shift":1
}
},
计算快速随机和慢速随机的指標 %K 和信号线 %D
使用三个"bucket_script"聚合，FK 用于快速 %K，FDSK 用于快速 %D 和慢速 %K，而SD 则用于慢速 %D。 使用参数" buckets_path"指定来自 Daily、MMin 和 MMax 的聚合结果。 然后，根据脚本中的公式计算快速 %K。 FDSK 是滑动窗口为3 的FK 简单移动平均值 ，SD 是滑动窗口为3 的FDSK 简单移动平均值。简单移动平均值使用函数MovingFunctions.unweightedAvg()来计算。 参数"shift"设置为 1 以包括最近的数据。
"FK": {
"bucket_script": {
"buckets_path": {"Daily": "Daily", "MMin": "MMin", "MMax": "MMax"}, "script": "100 * (params.Daily - params.MMin)/(params.MMax - params.MMin)"
}
},
"FDSK": {
"moving_fn": {
"script": "MovingFunctions.unweightedAvg(values)", "window": 3, "buckets_path": "FK", "shift": 1
}
},
"SD": {
"moving_fn": {
"script": "MovingFunctions.unweightedAvg(values)", "window": 3, "buckets_path": "FDSK", "shift": 1
}
},
识别 %K和 %D的交叉类型
a) 使用名为 FKFD_Diff 的"bucket_script"聚合，并配合参数"buckets_path"来指定 FK 和 FDSK 的值，以确定距离是正数还是负数。 如果 FK 高于 FDSK，则设置为 1。如果 FK 低于 FDSK，则设置为 -1。 如果它们相等，则设置为 0。 SKSD_Diff 聚合可以相同的方式定义。
"FKFD_Diff": {
"bucket_script": {
"buckets_path": {"FK": "FK", "FDSK": "FDSK"}, "script": "(params.FK - params.FDSK) > 0 ? 1 : ((params.FK - params.FDSK) == 0 ? 0 : -1)"
}
},
"SKSD_Diff": {
"bucket_script": {
"buckets_path": {"FDSK": "FDSK", "SD": "SD"}, "script": "(params.FDSK - params.SD) > 0 ? 1 : ((params.FDSK - params.SD) == 0 ? 0 : -1)"
}
},
b) 使用名为 F_Diff 的导数聚合，并配合参数"buckets_path"指定 FKFD_Diff 的值，以计算与前一个时间戳的值的差异。%K 和 %D 的交叉可能涉及一或两个交易日，所以F_Diff的值可以为-1、-2、1或2。在超卖区域需要关注-1 和 -2 的值，在超买区域需要关注1 和 2 的值。 S_Diff 聚合可以相同的方式定义为慢速随机指标。
"F_Diff": {
"derivative": {"buckets_path": "FKFD_Diff"}
},
"S_Diff": {
"derivative": {"buckets_path": "SKSD_Diff"}
},
c) 由于%K 和 %D 的交叉可能涉及一或两个交易日。 因此需要前一个交易日的数据。 使用名为 PRE_FK 的"moving_fn"聚合，并配合参数 window 为 1，参数"buckets_path"为 FK的简单移动总和值。简单移动总和值使用函数MovingFunctions.sum()来计算。PRE_FDSK 和 PRE_SD 聚合可以用相同的方式定义。
"PRE_FK": {
"moving_fn": {"script": "MovingFunctions.sum(values)", "window": 1, "buckets_path": "FK"}
},
"PRE_FDSK": {
"moving_fn": {"script": "MovingFunctions.sum(values)", "window": 1, "buckets_path": "FDSK"}
},
"PRE_SD": {
"moving_fn": {"script": "MovingFunctions.sum(values)", "window": 1, "buckets_path": "SD"}
},
d) 当%K和%D产生交叉时，原则上应该发生在超买或超卖区域。然而，在 Elasticsearch 服务器中计算实际交叉点甚不方便，因此为确保交叉在邻近正确的区域发生，两个交易日中必须有一日在超买或超卖区域。要确定交叉是否有效，聚合 F_Type 检查以下几个条件。 如果是卖出信号，则将 F_Type 设置为 1。如果是买入信号，则将 F_Type 设置为 -1。 否则，将 F_type 设置为 0。 S_Type 聚合可以用相同的方式定义。
- 在超买区域内
????????params.PRE_FK > 80 && params.PRE_FDSK > 80
- 在超卖区域内
????????params.PRE_FK < 20 && params.PRE_FDSK < 20
- 在超卖区域需要关注的交叉
????????params.F_Diff == -1 || params.F_Diff == -2
- 在超买区域需要关注的交叉
????????params.F_Diff == 1 || params.F_Diff == 2
- FK 在 FDSK 下方交叉
????????params..FK <= params..FDSK
- FK 在 FDSK 上方交叉
????????params..FK >= params..FDSK
"F_Type": {
"bucket_script": {
"buckets_path": {"F_Diff": "F_Diff", "FK": "FK", "FDSK": "FDSK", "SD": "SD", "PRE_FK": "PRE_FK", "PRE_FDSK": "PRE_FDSK", "PRE_SD": "PRE_SD"},
"script": "((params.F_Diff == -1 || params.F_Diff == -2) && (params.PRE_FK > 80 || params.FK > 80) && (params.PRE_FDSK > 80 || params.FDSK > 80) && params.FK <= params.FDSK) ? 1 : (((params.F_Diff == 1 || params.F_Diff == 2) && (params.PRE_FK < 20 || params.FK < 20) && (params.PRE_FDSK < 20 || params.FDSK < 20) && params.FK >= params.FDSK) ? -1 : 0)"
}
},
"S_Type": {
"bucket_script": {
"buckets_path": {"S_Diff": "S_Diff", "FDSK": "FDSK", "SD": "SD", "PRE_FDSK": "PRE_FDSK", "PRE_SD": "PRE_SD"},
"script": "((params.S_Diff == -1 || params.S_Diff == -2) && (params.PRE_FDSK > 80 || params.FDSK > 80) && (params.PRE_SD > 80 || params.SD > 80) && params.FDSK <= params.SD) ? 1 : (((params.S_Diff == 1 || params.S_Diff == 2) && (params.PRE_FDSK < 20 || params.FDSK < 20) && (params.PRE_SD < 20 || params.SD < 20)&& params.FDSK >= params.SD) ? -1 : 0)"
}
},
使用名为 S_Date 的"bucket_selector"聚合，并配合参数"buckets_path"为"DateStr”，“script"语句中指定存储桶的选择条件。 选择标准是日期在 2021 年 1月 1 日及之后的存储桶（以毫秒为单位指定纪元时间 1609459200000）。由于涉及当前收盘价，买入或卖出交易将顺延至下一个交易日。因此，所有的结果都会在python程序中上报和处理，根据策略买入或卖出交易。
"S_Date": {
"bucket_selector": {
"buckets_path": {"DateStr": "DateStr"}, "script": "params.DateStr >= 1609459200000L"
}
}
}
}
},
"from": 0, "size": 0
}
收集结果后，可以绘制如前图所示。
????????执行结果会发出买入或卖出信号； 然而，这些信号仅满足前述交易策略的第二种和第三种情况。 对于第一种和第四种情况，需要使用 Python 编程语言来编写程序。主程序包括四个部分。
- 读取两个命令行参数。 一个用于选定的代码，另一个用于包含使用 JSON 格式在 Elasticsearch REST API 请求正文中编写的交易策略的文件名称。
- 从 Elasticsearch 服务器获取数据。
- 解析响应数据并优化买入和卖出信号。
- 报告回测统计数据(为简单起见，利润并未扣除交易费用)。
主函数如下所示：
def main(argv):
inputfile, symbol, type = get_opt(argv)
resp = get_data(inputfile, symbol)
transactions = parse_data(resp, type)
report(transactions, type)
????????在本文中，仅展示了买卖信号细化的代码段。 读者可以进一步参考Gitee上的开源项目（Backtest_Stochastic）。 为确保每次只买入及只持有一股，并且在卖出持有的股票之前不发生交易，我们使用布尔变量"hold"来确保交易满足以下条件。
- 当hold为 False 时，买入信号(值等于 -1)被接受
- 当hold为True时，卖出信号(值等于1)被接受
????????函数parse_data()如下所示。 最后，交易数组transaction只包含有效信号。然而，这些信号将在report中按照交易策略处理买入或卖出。利润是使用复权单位净值差额计算。而买入价格和利润百分比则使用单位价格(unit_nav)计算。
# parse the response data and refine the buy/sell signal
def parse_data(resp, type):
result = json.loads(resp)
aggregations = result['aggregations']
if aggregations and 'Backtest_Stochastic' in aggregations:
Backtest_Stochastics = aggregations['Backtest_Stochastic']
transactions = []
hold = False
if Backtest_Stochastics and 'buckets' in Backtest_Stochastics:
for bucket in Backtest_Stochastics['buckets']:
transaction = {}
transaction['date'] = bucket['key_as_string']
transaction['Daily'] = bucket['Daily']['value']
# honor buy signal if there is no share hold
if bucket[type]['value'] == -1:
transaction['original'] = 'buy'
if not hold:
transaction['buy_or_sell'] = 'buy'
else:
transaction['buy_or_sell'] = 'hold'
hold = True
# honor sell signal if there is a share hold
elif bucket[type]['value'] == 1:
transaction['original'] = 'sell'
if hold:
transaction['buy_or_sell'] = 'sell'
else:
transaction['buy_or_sell'] = 'hold'
hold = False
# for other situations, just hold the action
else:
transaction['original'] = 'hold'
transaction['buy_or_sell'] = 'hold'
transactions.append(transaction)
return transactions
????????python程序提供交易策略的统计信息，包括整个买卖交易的"赢"和"输"。以下是对000803.OF运行快速随机交易策略后的结果。
购买次数: 2
卖出次数: 2
得胜次数: 1
亏损次数: 1
总利润: -0.13
平均购买价格: 3.24
利润百分率: -3.89%
????????从2021年01月01日到2021年05月31日，11只股票型基金运行简单型随机交易策略的结果汇总和展示如下表。结果表明，这个交易策略不一定可以获利。大多数交易者的建议不要根据单一指标进行交易。
????????下表为本次实验中，使用慢速随机交叉交易策略的结果，显示亏损于多于盈利。
????????下表为本次实验中，使用快速随机指标、慢速随机指标和相对强弱指数指标之间的收益。 在表格中显示相对强弱指数指标比其他两个指标具有更高的收益。然而，有一只基金在 RSI 中没有提供信号，但快速和慢速随机指标均获利。
????????下表是比较买入、卖出、赢和输的总次数。 相对强弱指数指标交叉交易策略有更好的表现。
备注:
- 感谢Tushare大数据开放社区提供相关数据及Gitee开源社区提供存储开源项目。
- 本文基于公开发布技术和研究观点，并不构成任何投资建议，读者在使用时须自行承担责任。
- 文中可能还存在疏漏和错误之处，恳请广大读者批评和指正。
- 作者的中文著作Elasticsearch 数据分析与实战应用（ISBN 978-7-113-27886-1号）将于2021 年 8月出版。
- 作者的英文著作Advanced Elasticsearch 7.0（ISBN 978-1-789-95775-4号）被bookauthority评为 2021 年最值得阅读的 4 本 Elasticsearch 新书之一。