1.移动平均（英语：Moving Average，MA），又称“移动平均线”简称均线，是技术分析中一种分析时间序列数据的工具。最常见的是利用股价、回报或交易量等变量计算出移动平均。移动平均可抚平短期波动，反映出长期趋势或周期。
- MA：移动平均，分为简单移动平均 SMA, 和加权移动平均 WMA；
移动平均（英语：Moving Average，MA），又称“移动平均线”简称均线，是技术分析中一种分析时间序列数据的工具。最常见的是利用股价、回报或交易量等变量计算出移动平均。移动平均可抚平短期波动，反映出长期趋势或周期。原本的意思是移动平均，由于我们将其制作成线形，所以一般称之为移动平均线，简称均线。它是将某一段时间的收盘价之和除以该周期。比如日线MA5指5天内的收盘价除以5。
移动平均线常用线有5天、10天、30天、60天、120天和240天的指标。其中，5天和10天的短期移动平均线。是短线操作的参照指标，称做日均线指标；30天和60天的是中期均线指标，称做季均线指标；120天、240天的是长期均线指标，称做年均线指标。对移动平均线的考查一般从几个方面进行。
移动平均线按时间周期长短分为：短期移动平均线，中期移动平均线，长期移动平均线；按计算方法分为：算术移动平均线，加权移动平均线，指数平滑移动平均线（EMA）。
- MACD：由MA演化而来；
指数平滑异同移动平均线（Moving Average Convergence / Divergence，MACD）是股票交易中一种常见的技术分析工具，由Gerald Appel于1970年代提出，用于研判股票价格变化的强度、方向、能量，以及趋势周期，以便把握股票买进和卖出的时机。MACD指标由一组曲线与图形组成，通过收盘时股价或指数的快变及慢变的指数移动平均值（EMA）之间的差计算出来。“快”指更短时段的EMA，而“慢”则指较长时段的EMA，最常用的是12及26日EMA。
MACD称为指数平滑异同平均线，是从双指数移动平均线发展而来的，由快的指数移动平均线（EMA）减去慢的指数移动平均线，MACD的意义和双移动平均线基本相同，但阅读起来更方便。当MACD从负数转向正数，是买的信号。当MACD从正数转向负数，是卖的信号。当MACD以大角度变化，表示快的移动平均线和慢的移动平均线的差距非常迅速的拉开，代表了一个市场大趋势的转变。
- 特点
1）追踪趋势。
追踪趋势。注意价格的趋势，并追随这个趋势，不轻易放弃。如果从股价的图表中能够找出上升或下降趋势线，那么，MA的曲线将保持与趋势线方向一致，能消除中间股价在这个过程中出现的起伏。原始数据的股价图表不具备这个保持追踪趋势的特性。
2）稳定性。
通常愈长期的移动平均线，愈能表现安定的特性，即移动平均线不轻易往上往下，必须股价涨势真正明朗了，移动平均线才会往上延伸，而且经常股价开始回落之初，移动平均线却是向上的，等到股价下滑显著时，才见移动平均线走下坡，这是移动平均线最大的特色。愈短期的移动平均线，安定性愈差，愈长期移动平均线，安定性愈强，但也因此使得移动平均线有延迟反应的特性。
3）滞后性。
在股价原有趋势发生反转时，由于MA的追踪趋势的特性，MA的行动往往过于迟缓，调头速度落后于大趋势。这是MA的一个极大的弱点。等MA发出反转信号时，股价调头的深度已经很大了。
4）助涨助跌性。
助涨助跌性。当股价突破了MA时，无论是向上突破还是向下突破，股价有继续向突破方面再走一程的愿望，这就是MA的助涨助跌性。
5）依靠性。
6）支撑线和压力线的特性
- 特征分析
1.多头稳定上升
当多头市场进入稳定上升时期,10MA、21MA、68MA向右上方推升,且三线多头排列(排列顺序自上而下分别为10MA、21MA、68MA)、略呈平行状。
2.技术回档
当10MA由上升趋势向右下方拐头而下,而21MA仍然向上方推升时,揭示此波段为多头市场中的技术回档,涨势并未结束。
3.由空转多
股市由空头市场进入多头市场时,10MA首先由上而下穿越K线图(注意是K线图),处于k线图的下方(即股价站在10MA之上),过几天21MA、68MA相继顺次,由上往下穿越K线图(既股价顺次站在21MA、68MA之上)。
4.股价盘整
股价盘整时10MA与21MA交错在一起,若时间拉长68MA也会粘合在一起。
5.盘高与盘低
股价处于盘局时若10MA往右上方先行突破上升,则后市必然盘高;若10MA往右下方下降时,则后市必然越盘越低。
6.空头进入尾声
空头市场中,若68MA能随10MA于21MA之后,由上而下贯穿K线图(既股价站在68MA之上),则后市会有一波强劲的反弹,甚至空头市场至此已接近尾声。
7.由多转空
若21MA随10MA向右下方拐头而下,68MA也开始向右下方反转时,表示多头市场既将结束,空头市场既将来临。
8.跌破10MA
当市场由多头市场转入空头市场时,10MA首先由下往上穿越K线图,到达K线图的上方(股价跌破10MA),过几天30MA、68MA相继顺次由下往上穿越K线图,到达K线图的上方。
9.移动平均线依次排列
空头市场移动平均线均在K线图之上,且排列顺序从上而下依次是68MA、21MA、10MA。
10.反弹开始
空头市场中,若移动10MA首先从上而下穿越K线图时(K线图在上方,10MA在下方)既股价站在10MA之上,是股价在空头市场反弹的先兆。
11.反弹趋势增强
空头市场中,若21MA也继10MA之后,由上而下穿越K线图,且10MA位于21MA之上(既股价站在21MA之上,10MA、21MA多头排列),则反弹趋势将转强。
12.深幅回档
若21MA随10MA向右下方拐头而下,68MA仍然向右上方推升时,揭示此波段为多头市场中的深幅回档。应以持币观望或放空的策略对应。
- 格兰维尔法则
1、移动平均线从下降逐渐走平且略向上方抬头，而股价从移动平均线下方向上方突破，为买进信号。
2、股价位于移动平均线之上运行，回档时未跌破移动平均线后又再度上升时为买进时机。
3、股价位于移动平均线之上运行，回档时跌破移动平均线，但短期移动平均线继续呈上升趋势，此时为买进时机。
4、股价位于移动平均线以下运行，突然暴跌，距离移动平均线太远，极有可能向移动平均线靠近(物极必反，下跌反弹)，此时为买进时机。
5、股价位于移动平均线之上运行，连续数日大涨，离移动平均线愈来愈远，说明内购买股票者获利丰厚，随时都会产生获利回吐的卖压，应暂时卖出持股。
6、移动平均线从上升逐渐走平，而股价从移动平均线上方向下跌破移动平均线时说明卖压渐重，应卖出所持股票。
7、股价位于移动平均线下方运行，反弹时未突破移动平均线，且移动平均线跌势减缓，趋于水平后又出现下跌趋势，此时为卖出时机。
8、股价反弹后在移动平均线上方徘徊，而移动平均线却继续下跌，宜卖出所持股票。算法
import pandas as pd
DataAPI.settings.cache_enabled = Truedef cal_ema(data, shortNumber, longNumber):
"""计算N日EMA指数平均值，注意处理停盘情况
EMA[i] = EMA[i-1] * (long - 1)/(long + 1) + closePrice * 2 / (long + 1)
"""
ema_short = [data['closePrice'][0]] * len(data)
ema_long = [data['closePrice'][0]] * len(data)
for i in range(1, len(data)):
if data['turnoverVol'][i] == 0:
ema_short[i] = ema_short[i-1]
ema_long[i] = ema_long[i-1]
else:
ema_short[i] = ema_short[i-1] * (shortNumber-1)/(shortNumber+1) + data['closePrice'][i] * 2/(shortNumber+1)
ema_long[i] = ema_long[i-1] * (longNumber-1)/(longNumber+1) + data['closePrice'][i] * 2/(longNumber+1)
data['EMA_' + str(shortNumber)] = ema_short
data['EMA_' + str(longNumber)] = ema_long
return data
def cal_diff(ticker, listDate, shortNumber=12, longNumber=26):
"""计算DIFF偏移值
DIFF = EMA(short) - EMA(long)
"""
data = DataAPI.MktEqudAdjGet(ticker=ticker, beginDate=listDate.replace('-',''),
field=["secShortName", "tradeDate", "closePrice", "turnoverVol"])
ema_data = cal_ema(data, shortNumber, longNumber)
ema_data['DIFF'] = ema_data["EMA_" + str(shortNumber)] - ema_data["EMA_" + str(longNumber)]
return ema_data
def cal_dea(data, n=9):
"""计算DEA差离平均值
DEA[i] = DEA[i-1] * (n-1) / (n+1) + DIFF[i] * 2 / (n+1)
"""
dea = [data['DIFF'][0]] * len(data)
for i in range(1, len(data)):
if data['turnoverVol'][i] == 0:
dea[i] = dea[i-1]
else:
dea[i] = dea[i-1] * (n-1)/(n+1) + data['DIFF'][i] * 2/(n+1)
data['DEA'] = dea
return data
def cal_macd(ticker, listDate=None):
"""计算MACD指数平滑平均值
MACD = 2 * (DIFF - DEA)
"""
if not listDate:
listDate = DataAPI.SecIDGet(assetClass='E', ticker=ticker, field=['listDate']).listDate[0]
diff_data = cal_diff(ticker, listDate)
macd_data = cal_dea(diff_data)
macd_data['MACD'] = 2 * (macd_data['DIFF'] - macd_data['DEA'])
return macd_data测试
df = cal_macd('000002')
df.tail()df_open = df[df.turnoverVol != 0]df_open.tail()
作图
from lib import reportdf_open_tail = df_open.tail(100)# report library: https://uqer.datayes.com/community/share/56a5b221228e5b2047d916d9ax = df_open_tail.plot(x=['tradeDate'], y=['closePrice'], figsize=(20, 6))report.fig_style(ax, ['closePrice'])ax = df_open_tail.plot(x=['tradeDate'], y=['MACD'], kind='bar', figsize=(20, 6), color='r')df_open_tail['tmp'] = [min(i, 0) for i in df_open_tail.MACD]df_open_tail.plot(y=['tmp'], kind='bar', color='g', ax=ax)report.fig_style(ax, ['MACD'])
2.ADX，全称Average Directional Index，是一种用于衡量资产行情是否存在趋势的技术指标
ADX指标
- ADX，全称Average Directional Index，是一种用于衡量资产行情是否存在趋势的技术指标
- ADX的计算方法十分复杂，涉及到+DM, -DM, TR, +DI, -DI等很多中间变量：
- 首先计算+DM和-DM，即所谓上升动向与下降动向，公式为
- +DMt=t日最高价 – (t-1)日最高价−DMt=(t-1)日最低价 – t日最低价
- 其次我们来计算TR，即所谓的真实波幅：TRt=max(|t日最高−t日最低|,|t日最高−(t−1)日收盘|,|t日最低−(t−1)日收盘|)
- 然后利用+DM,-DM,TR来计算DI：±DMt=∑N−1k=0±DMt−k/∑N−1k=0TRt−k×100，这里的N是参数，一般取14
- 然后我们就可以计算ADX和ADXR了：
- ADXt=MA((+DIt)−(−DIt)(+DIt)+(−DIt)×100,N)
- 回测结果，首先使用动态Universe消除survival bias，ADX可以在几乎为1的beta下带来一定的alpha，但是绝对水平较低，需要和其他因子配合使用
import numpy as np
import pandas as pd
start = '20120101'
end = '20160901'
benchmark = 'HS300'
universe = DynamicUniverse('HS300')
capital_base = 1000000
freq = 'd'
refresh_rate = 5
adx_threshold = 40 # 判断是否存在趋势的阈值
rtn_threshold = 0.05 # 判断累计收益的变化阈值
# 定义并注册Signal模块
def initialize(account):
account.signal_generator = SignalGenerator(Signal('ADX'))
def handle_data(account):
history = account.get_attribute_history('closePrice', 14) # 由于计算ADX用的周期是14
returns = {sec: history[sec][-1] / history[sec][0] - 1 for sec in account.universe}
avgret = sum(returns.values()) / len(returns)
up_down = {sec: returns[sec] > 0 for sec in account.universe}
up_percent = 1. * sum(up_down.values()) / len(up_down)
buylist = []
cash = account.cash
for sec in account.universe:
if account.signal_result['ADX'].get(sec, 0) > adx_threshold and returns[sec] > rtn_threshold and sec not in account.security_position: # 存在上涨趋势
buylist.append(sec)
elif account.signal_result['ADX'].get(sec, 0) > adx_threshold and returns[sec] < -rtn_threshold and sec in account.security_position: # 存在下跌趋势
order_to(sec, 0) # 全部卖出
cash += account.security_position[sec] * account.reference_price[sec] # 估计买入金额
for sec in buylist:
order(sec, cash / len(buylist) / account.reference_price[sec])3.CMO(Chande Momentum Oscillator)动量震荡指标是由Tushar S. Chande提出的类似于RSI的指标
指标介绍
- **CMO(Chande Momentum Oscillator)**动量震荡指标是由Tushar S. Chande提出的类似于RSI的指标
- CMOn是一个n天滚动指标，在这n天中的第i天计算每天的 收盘价 – 前收盘价 ，如果为正则赋给upi（dni为0），为负则将绝对值赋给dni（upi为0）
- 其计算公式为：
CMOn=∑ni=1upi−∑ni=1dni∑ni=1upi+∑ni=1dni∗100
策略思路
- 计算上证50成分股当中所有股票过去n天的CMO
- CMO大于0时买入，小于0时卖出
- 根据一定的调仓原则进行调仓，细节见代码
可进一步挖掘的点
- 考虑CMO的形态，如上/下穿0线作为买卖信号
- 扩大股票池范围，观察CMO与股票池的关系，比如区分大小盘股观察CMO的有效性
- 股票权重的分配方式
- 其他调仓原则
import numpy as npstart = '2010-01-01'end = '2015-06-20'benchmark = 'SH50'universe = set_universe('SH50')capital_base = 1000000window = 35 # 参数，CMO指标计算周期def initialize(account): passdef handle_data(account): clp = account.get_attribute_history('closePrice', window) prc = account.get_attribute_history('preClosePrice', window) p = account.referencePrice # 计算CMO CMO = {} for s in account.universe: diff = clp[s] - prc[s] u = sum(n for n in diff if n > 0) d = sum(-n for n in diff if n < 0) if u + d == 0: continue CMO[s] = (u - d) / (u + d) * 100 # 根据CMO卖出目前持有股票 v = account.cash for s,a in account.valid_secpos.items(): if CMO.get(s, 0) < 0 and s in account.universe: order_to(s, 0) v += a * p[s] # 根据CMO确定买入列表 buylist = [] for s in account.universe: if CMO.get(s, 0) > 0 and not np.isnan(p[s]) and s not in account.valid_secpos: buylist.append(s) # 根据买入列表和可用现金买入股票 if v > account.referencePortfolioValue * 0.33: # 为了避免调仓过于频繁，仅当可用现金超过账户市值1/3时买入 for s in buylist: order(s, v / len(buylist) / p[s])
#####上面的策略实现了策略思路所表述的意思，使用了非常简单的调仓原则，其表现还不错
#####但是，其中的关键参数 window 为什么设置为 35 呢？
#####这当然不是拍脑袋拍出来的，而是通过参数调试出来的:
start = '2010-01-01'
end = '2015-06-20'
benchmark = 'SH50'
universe = set_universe('SH50')
capital_base = 1000000.
sim_params = quartz.sim_condition.env.SimulationParameters(start, end, benchmark, universe, capital_base)
idxmap_all, data_all = quartz.sim_condition.data_generator.get_daily_data(sim_params)import numpy as np
def initialize(account):
pass
def handle_data(account):
clp = account.get_attribute_history('closePrice', window)
prc = account.get_attribute_history('preClosePrice', window)
p = account.referencePrice
CMO = {}
for s in account.universe:
diff = clp[s] - prc[s]
u = sum(n for n in diff if n > 0)
d = sum(-n for n in diff if n < 0)
if u + d == 0: continue
CMO[s] = (u - d) / (u + d) * 100
buylist = []
for s in account.universe:
if CMO.get(s, 0) > 0 and not np.isnan(p[s]) and s not in account.valid_secpos:
buylist.append(s)
v = account.cash
for s,a in account.valid_secpos.items():
if CMO.get(s, 0) < 0 and s in account.universe:
order_to(s, 0)
v += a * p[s]
if v > account.referencePortfolioValue * 0.33:
for s in buylist:
order(s, v / len(buylist) / p[s])
print 'window annualized_return sharpe max_drawdown'
for window in range(10, 51, 5):
strategy = quartz.sim_condition.strategy.TradingStrategy(initialize, handle_data)
bt_test, acct = quartz.quick_backtest(sim_params, strategy, idxmap_all, data_all)
perf = quartz.perf_parse(bt_test, acct)
print ' {0:2d} {1:>7.4f} {2:>7.4f} {3:>7.4f}'.format(window, perf['annualized_return'], perf['sharpe'], perf['max_drawdown'])从上面的调试结果中可以看到，当 window = 35 时，夏普和最大回撤相对而言最好，因此有了最上面的那个策略
然而调试完了window，这个策略就没有优化空间了吗？不，我们还可以根据这个策略的表现来分析一下这个策略的缺陷在哪里，并加以改进
因为该策略的调仓原则是买入所有产生的信号，并没有对持仓进行限制，这会造成两个方面的影响：
- 仓位中的股票可能会有很多只，这样资金会比较分散，削弱信号的效果
- 如果买入信号比较少，而卖出信号比较多的话，现金的利用率会比较低
那么到底是否存在上述问题呢？我们可以通过最大持仓数量和现金走势来加以判断
x = map(len, bt['security_position'].tolist())max(x)42
bt.cash.plot()从上面的两个cell中可以看出这两个问题还是比较明显的。为了解决这两个问题，我们对策略进行优化：一是限制最大持仓位10只股票，二是每次卖出的现金都平均分配给目前仓位中的股票和即将买入的股票
import numpy as npfrom heapq import nlargeststart = '2010-01-01'end = '2015-06-20'benchmark = 'SH50'universe = set_universe('SH50')capital_base = 1000000max_n = 10 # 参数，最大持仓数量 window = 15 # 参数，CMO指标计算周期def initialize(account): passdef handle_data(account): clp = account.get_attribute_history('closePrice', window) prc = account.get_attribute_history('preClosePrice', window) p = account.referencePrice # 计算CMO CMO = {} for s in account.universe: diff = clp[s] - prc[s] u = sum(n for n in diff if n > 0) d = sum(-n for n in diff if n < 0) if u + d == 0: continue CMO[s] = (u - d) / (u + d) * 100 # 根据CMO卖出目前持有股票 n = len(account.valid_secpos) sellist = [] for s,a in account.valid_secpos.items(): if CMO.get(s, 0) < 0 and s in account.universe: order_to(s, 0) n -= 1 sellist.append(s) if n >= max_n: # 如果超过最大持仓，则不买入 return # 根据CMO确定买入列表 buylist = [] for s in account.universe: if CMO.get(s, 0) > 0 and not np.isnan(p[s]) and s not in account.valid_secpos: buylist.append(s) # 根据最大持仓数量确定买入列表数量，按CMO排序选较大的部分 if len(buylist) + n > max_n: buylist = nlargest(max_n - n, buylist, key=CMO.get) # 将资金重新分配到新买入的与已持有的股票中 buylist += [s for s in account.valid_secpos if s not in sellist] amount = {} for s in buylist: amount[s] = account.referencePortfolioValue / len(buylist) / p[s] - account.valid_secpos.get(s, 0) # 根据应调数量买卖股票，先卖出后买入 for s in sorted(amount, key=amount.get): order(s, amount[s])
import numpy as npfrom heapq import nlargestmax_n = 10 # 参数，最大持仓数量 def initialize(account): passdef handle_data(account): clp = account.get_attribute_history('closePrice', window) prc = account.get_attribute_history('preClosePrice', window) p = account.referencePrice # 计算CMO CMO = {} for s in account.universe: diff = clp[s] - prc[s] u = sum(n for n in diff if n > 0) d = sum(-n for n in diff if n < 0) if u + d == 0: continue CMO[s] = (u - d) / (u + d) * 100 # 根据CMO卖出目前持有股票 n = len(account.valid_secpos) sellist = [] for s,a in account.valid_secpos.items(): if CMO.get(s, 0) < 0 and s in account.universe: order_to(s, 0) n -= 1 sellist.append(s) if n >= max_n: # 如果超过最大持仓，则不买入 return # 根据CMO确定买入列表 buylist = [] for s in account.universe: if CMO.get(s, 0) > 0 and not np.isnan(p[s]) and s not in account.valid_secpos: buylist.append(s) # 根据最大持仓数量确定买入列表数量，按CMO排序选较大的部分 if len(buylist) + n > max_n: buylist = nlargest(max_n - n, buylist, key=CMO.get) # 将资金重新分配到新买入的与已持有的股票中 buylist += [s for s in account.valid_secpos if s not in sellist] amount = {} for s in buylist: amount[s] = account.referencePortfolioValue / len(buylist) / p[s] - account.valid_secpos.get(s, 0) # 根据应调数量买卖股票，先卖出后买入 for s in sorted(amount, key=amount.get): order(s, amount[s])print 'window annualized_return sharpe max_drawdown'for window in range(10, 51, 5): strategy = quartz.sim_condition.strategy.TradingStrategy(initialize, handle_data) bt_test, acct = quartz.quick_backtest(sim_params, strategy, idxmap_all, data_all) perf = quartz.perf_parse(bt_test, acct) print ' {0:2d} {1:>7.4f} {2:>7.4f} {3:>7.4f}'.format(window, perf['annualized_return'], perf['sharpe'], perf['max_drawdown'])从上面的图表可以看出其表现相比最初的策略有了不少改善。其中 window = 15 是最适合目前调仓原则的参数。
但是这些优化的初衷是为了解决股票数量和资金利用率的问题，我们仍然通过最大持仓数量和现金走势来判断
x = map(len, bt['security_position'].tolist())max(x)10
bt.cash.plot()以上是对CMO这个技术指标进行的一些简单的回测，并且针对策略本身的特点进行了一定的优化。在最前面列出了一些可挖掘的点，如果想进行更深入的研究还是有很多东西可以做的。
4.简易波动指标（EMV），是为数不多的考虑价量关系的技术指标。它刻画了股价在下跌的过程当中，由于买气不断的萎靡退缩，致使成交量逐渐的减少，EMV 数值也因而尾随下降，直到股价下跌至某一个合理支撑区，捡便宜货的买单促使成交量再度活跃，EMV 数值于是作相对反应向上攀升，当EMV 数值由负值向上趋近于零时，表示部分信心坚定的资金，成功的扭转了股价的跌势，行情不断反转上扬，并且形成另一次的买进讯号。
简易波动指标（EMV），是为数不多的考虑价量关系的技术指标。它刻画了股价在下跌的过程当中，由于买气不断的萎靡退缩，致使成交量逐渐的减少，EMV 数值也因而尾随下降，直到股价下跌至某一个合理支撑区，捡便宜货的买单促使成交量再度活跃，EMV 数值于是作相对反应向上攀升，当EMV 数值由负值向上趋近于零时，表示部分信心坚定的资金，成功的扭转了股价的跌势，行情不断反转上扬，并且形成另一次的买进讯号。
计算方法：
第一步
MID=TH+TL2 −YH+YL2
这里TH 为当天最高价，TL 为当天最低价，YH 为前日最高价，YL 为前日最低价。MID > 0意味着今天的平均价高于昨天的平均价。
第二步
BRO=VOLH−L
其中VOL代表交易量，H、L代表同一天的最高价与最低价
第三步
EM=MIDBRO
第四步
EMV = EM的N日简单移动平均
第五步
MAEMV = EMV的M日简单移动平均
def emv(stk_list,current_date,N=14):
cal = Calendar('China.SSE')
period = '-' + str(N+1) + 'B'
begin_date = cal.advanceDate(current_date,period,BizDayConvention.Unadjusted)
end_date = cal.advanceDate(current_date,'-1B',BizDayConvention.Unadjusted)
eq_emv = {}
eq_mid = {}
eq_bro = {}
eq_Market = DataAPI.MktEqudAdjGet(secID=stk_list,beginDate=begin_date.strftime('%Y%m%d'),endDate=end_date.strftime('%Y%m%d'),field=['secID','highestPrice','lowestPrice','turnoverVol'],pandas="1")
avaiable_list = eq_Market['secID'].drop_duplicates().tolist()
eq_Market.set_index('secID',inplace=True)
for stk in avaiable_list:
if len(eq_Market.ix[stk]) == (N+1):
eq_mid[stk] = (np.array(eq_Market.ix[stk]['highestPrice'][1:] + eq_Market.ix[stk]['lowestPrice'][1:]) - np.array(eq_Market.ix[stk]['highestPrice'][:-1] + eq_Market.ix[stk]['lowestPrice'][:-1]))/2
eq_bro[stk] = np.array(eq_Market.ix[stk]['turnoverVol'][1:])/np.array(eq_Market.ix[stk]['highestPrice'][1:] + eq_Market.ix[stk]['lowestPrice'][1:])
eq_emv[stk] = np.mean(eq_mid[stk]/eq_bro[stk])
return eq_emvdef maemv(stk_list,current_date,N=14):
cal = Calendar('China.SSE')
period = '-' + str(N+1) + 'B'
end_date = cal.advanceDate(current_date,'-1B',BizDayConvention.Unadjusted)
start_date = cal.advanceDate(current_date,period,BizDayConvention.Unadjusted)
timeSeries = cal.bizDatesList(start_date, end_date)
eq_maemv = {}
#初始化eq_maemv字典
eq_emv = emv(stk_list,end_date,N)
for stk in eq_emv:
eq_maemv[stk] = 0
#仅调用N次emv函数
for i in xrange(len(timeSeries)):
eq_emv = emv(stk_list,timeSeries[i],N)
for stk in eq_emv:
eq_maemv[stk] = eq_maemv[stk] + eq_emv[stk]
for stk in eq_maemv:
eq_maemv[stk] = eq_maemv[stk]/N
return eq_maemvEMV指标基本用法
EMV 在0 以下表示弱势，在0 以上表示强势；EMV 由负转正应买进，由正转负应卖出。
import numpy as np
import pandas as pd
from CAL.PyCAL import *
start = '2012-08-01' # 回测起始时间
end = '2015-08-01' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = set_universe('HS300') # 证券池，支持股票和基金
capital_base = 1000000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 10 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
cal = Calendar('China.SSE')
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
eq_emv = emv(account.universe,account.current_date,N=14)
buylist = []
for stk in eq_emv:
if eq_emv[stk] > 0:
buylist.append(stk)
for stk in account.valid_secpos:
if stk not in eq_emv or eq_emv[stk] <= 0:
order_to(stk,0)
else:
if stk not in buylist[:]:
buylist.append(stk)
for stk in buylist:
order_to(stk,account.referencePortfolioValue/account.referencePrice[stk]/len(buylist))EMV结合MAEMV使用
EMV 上穿MAEMV 则买入，EMV 下穿MAEMV 则卖出。
import numpy as np
import pandas as pd
from CAL.PyCAL import *
start = '2012-08-01' # 回测起始时间
end = '2015-08-01' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = set_universe('HS300') # 证券池，支持股票和基金
capital_base = 1000000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 10 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
cal = Calendar('China.SSE')
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
eq_emv = emv(account.universe,account.current_date,14)
eq_maemv = maemv(account.universe,account.current_date,14)
buylist = []
for stk in eq_emv:
try:
if eq_emv[stk] > eq_maemv[stk]:
buylist.append(stk)
except:
pass
for stk in account.valid_secpos:
if stk not in eq_emv or stk not in eq_maemv or eq_emv[stk] <= eq_maemv[stk]:
order_to(stk,0)
else:
if stk not in buylist[:]:
buylist.append(stk)
for stk in buylist:
order_to(stk,account.referencePortfolioValue/account.referencePrice[stk]/len(buylist))5.KDJ是以最高价、最低价及收盘价为基本数据进行计算，得出的K值、D值和J值分别在指标的坐标上形成的一个点，连接无数个这样的点位，就形成一个完整的、能反映价格波动趋势的KDJ指标。它主要是利用价格波动的真实波幅来反映价格走势的强弱和超买超卖现象，在价格尚未上升或下降之前发出买卖信号的一种技术工具。
其实就是KDJ的取值情况，收益较MACD差点，作为摆动指标，在趋势明显的状态下表现较差，较MACD趋势指标可以比较容易看出来。
#######常用函数
def Date_type_preceding(date,period):
dt = Date.fromDateTime(date) #从python标准库datetime类型转化为cal的date类型
cal = Calendar('China.SSE') #上海证券交易所交易日历
lastDay = cal.advanceDate(dt,period,BizDayConvention.Preceding) #在cal日历下当前日期的前一工作日
day = lastDay.strftime('%Y%m%d')
return day#######state函数
def Get_KDJ_state(date,universe):
Factor = DataAPI.MktStockFactorsOneDayGet(tradeDate=date,secID=universe,field=['secID','KDJ_K','KDJ_D'],pandas="1")
Factor.set_index('secID',inplace=True)#转换index为ticker
Factor = Factor.dropna()
KDJ_temp_on = Factor[Factor.KDJ_K>Factor.KDJ_D]#取K在D上的股票
KDJ_temp_on['KDJ_state'] = 1#加入signal权重
KDJ_temp_under = Factor[Factor.KDJ_K<Factor.KDJ_D]
KDJ_temp_under['KDJ_state'] = -1
KDJ_temp = pd.concat([KDJ_temp_on,KDJ_temp_under],axis = 0)
KDJ = KDJ_temp.dropna()
KDJ = KDJ.sort_index()
return KDJ.KDJ_state#serialsfrom CAL.PyCAL import *
import numpy as np
import pandas as pd
start = '2014-01-01' # 回测起始时间
end = u'' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = set_universe('A') # 证券池，支持股票和基金
capital_base = 100000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 1 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
commission = Commission(buycost=0.0008, sellcost=0.0008)#调仓成本
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
yesterday = Date_type_preceding(account.current_date,'-1B')
KDJ_state = Get_KDJ_state(yesterday,account.universe)
KDJ_state = pd.DataFrame(KDJ_state)
buy = KDJ_state[KDJ_state.KDJ_state>0]#买卖list赋值
sell = KDJ_state[KDJ_state.KDJ_state<0]
buy = buy.dropna()
buy = buy.index
sell = sell.dropna()
sell = sell.index
for stk in sell:#遍历现有头寸
if stk in account.avail_secpos:
order_pct_to(stk,0) #卖出股票
for stk in buy:
order_pct_to(stk,0.01)#买入股票，粗暴简陋的分散投资方式，可减少回撤率
return
6.阿隆指标（Aroon）是由图莎尔·钱德（Tushar Chande)1995 年发明的，它通过计算自价格达到近期最高值和最低值以来所经过的期间数，帮助投资者预测证券价格从趋势到区域、区域或反转的变化。
一、阿隆指标(Aroon)简介
阿隆指标（Aroon）是由图莎尔·钱德（Tushar Chande)1995 年发明的，它通过计算自价格达到近期最高值和最低值以来所经过的期间数，帮助投资者预测证券价格从趋势到区域、区域或反转的变化。在技术分析领域中，有一个说法，一个指标使用的人越多，其效力越低。这个技术指标还挺冷门的，我们一同来看看它的效果。
from CAL.PyCAL import *
import numpy as np
import pandas as pd
from pandas import DataFrame
from heapq import nlargest
from heapq import nsmallest二、Aroon计算方法
Aroon指标分为两个具体指标，分别AroonUp和AroonDown。其具体计算方式为：
- AroonUp = [(计算期天数-最高价后的天数)/计算期天数]*100
- AroonDown = [(计算期天数-最低价后的天数)/计算期天数]*100
- AroonOsc = AroonUp – AroonDown
计算期天数通常取20天
def aroonUp(account,timeLength=20):
#运用heapq包的nlargest函数，可以轻松获得：计算期天数-最高价后的天数
eq_AroonUp = {}
history = account.get_attribute_history('closePrice',timeLength)
for stk in account.universe:
priceSeries = pd.Series(history[stk])
eq_AroonUp[stk] = (nlargest(1,range(len(priceSeries)),key=priceSeries.get)[0]+1)*100/timeLength # eq_AroonUp[stk]范围在[5,100]之间
return eq_AroonUpdef aroonDown(account,timeLength=20):
#运用heapq包的nsmallest函数，可以轻松获得:计算期天数-最低价后的天数
eq_AroonDown = {}
history = account.get_attribute_history('closePrice',timeLength)
for stk in account.universe:
priceSeries = pd.Series(history[stk])
eq_AroonDown[stk] = (nsmallest(1,range(len(priceSeries)),key=priceSeries.get)[0]+1)*100/timeLength # eq_AroonDown[stk]范围在[5,100]之间
return eq_AroonDown三、Aroon指标的基本用法
- 当AroonUp指标向下跌破50 时，表示向上的趋势正在失去动力；当AroonDown指标向下跌破50时，表示向下的趋势正在失去动力；如果两个指标都在低位，表示股价没有明确的趋势；如果指标在70 以上，表示趋势十分强烈；如果在30 以下，表明相反的趋势正在酝酿。通常来说，AroonOsc在0附近时，是典型的无趋势特征，股票处于盘整阶段。
- 参考研报《技术指标系列（三）——加入“二次确认”的AROON 阿隆优化指标》中的方法，我们买入AroonOsc > 50的股票。
start = '2009-08-01' # 回测起始时间
end = '2015-08-31' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = set_universe('HS300') # 证券池，支持股票和基金
capital_base = 100000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 10 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
eq_AroonUp = aroonUp(account,20)
eq_AroonDown = aroonDown(account,20)
buyList = []
for stk in account.valid_secpos:
order_to(stk, 0)
for stk in account.universe:
if eq_AroonUp[stk] - eq_AroonDown[stk] > 50:
buyList.append(stk)
for stk in buyList[:]:
if stk not in account.universe or account.referencePrice[stk] == 0 or np.isnan(account.referencePrice[stk]):
buyList.remove(stk)
for stk in buyList:
order(stk, account.referencePortfolioValue/account.referencePrice[stk]/len(buyList))
可以看出，策略在股市处于震荡市和牛市中，表现很好；而在熊市和暴跌中，表现的非常差，最大回撤很大。这从阿隆指标的构造中，就可以理解，阿隆指标是一个跟踪趋势的指标，在震荡市和牛市中，都能精选出股票，超越指数；然而在暴跌中，处于上升趋势的股票可能跌的更惨，倾巢之下，焉有完卵。
四、运用Aroon指标来择时
前文说到阿隆指标是一个跟踪趋势的指标，既然如此，我们为什么不把它用来择时呢？
def aroonIndex(account,timeLength=20):
#构建指数阿隆指标
indexSeries = pd.Series(account.get_symbol_history('benchmark', timeLength)['closeIndex'])
indexAronUp = (nlargest(1,range(len(indexSeries)),key=indexSeries.get)[0]+1)*100/timeLength
indexAronDown = (nsmallest(1,range(len(indexSeries)),key=indexSeries.get)[0]+1)*100/timeLength
indexOsc = indexAronUp - indexAronDown
return indexOsc当indexOsc > 0时，我们大致认为现在的市场环境没有那么差，可以考虑开仓，编写如下策略。
start = '2009-08-01' # 回测起始时间
end = '2015-08-31' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = set_universe('HS300') # 证券池，支持股票和基金
capital_base = 100000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 10 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
eq_AroonUp = aroonUp(account,20)
eq_AroonDown = aroonDown(account,20)
index_osc = aroonIndex(account,20)
buyList = []
for stk in account.valid_secpos:
order_to(stk, 0)
if index_osc > 0:
for stk in account.universe:
if eq_AroonUp[stk] - eq_AroonDown[stk] > 50:
buyList.append(stk)
for stk in buyList[:]:
if stk not in account.universe or account.referencePrice[stk] == 0 or np.isnan(account.referencePrice[stk]):
buyList.remove(stk)
for stk in buyList:
order(stk, account.referencePortfolioValue/account.referencePrice[stk]/len(buyList))可以看出运用阿隆指标来择时的效果还是不错的，震荡市能跑赢指数，牛市的收益基本可以吃到，暴跌也几乎完美的规避了！缺点就是最大回测还是偏大，可以考虑让条件更严格，让indexOsc > 50。
start = '2009-08-01' # 回测起始时间
end = '2015-08-31' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = set_universe('HS300') # 证券池，支持股票和基金
capital_base = 100000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 10 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
eq_AroonUp = aroonUp(account,20)
eq_AroonDown = aroonDown(account,20)
index_osc = aroonIndex(account,20)
buyList = []
for stk in account.valid_secpos:
order_to(stk, 0)
if index_osc > 50:
for stk in account.universe:
if eq_AroonUp[stk] - eq_AroonDown[stk] > 50:
buyList.append(stk)
for stk in buyList[:]:
if stk not in account.universe or account.referencePrice[stk] == 0 or np.isnan(account.referencePrice[stk]):
buyList.remove(stk)
for stk in buyList:
order(stk, account.referencePortfolioValue/account.referencePrice[stk]/len(buyList))将择时条件设置更严格后，最大回撤果然有所下降，但年化收益率也有大幅下降。从回测图形中，也可以明显看到，指标具有很强的滞后性，往往是指数开始涨了一段时间，策略才开始开仓买入。将indexOsc条件设置的越严格，滞后性表现的就越明显，这样虽然可以提高正确率，减小最大回撤，但有许多收益也错过了。
7.布林线指标，即BOLL指标，其英文全称是“Bollinger Bands”，布林线(BOLL)由约翰 布林先生创造，其利用统计原理，求出股价的标准差及其信赖区间，从而确定股价的波动范围及未来走势，利用波带显示股价的安全高低价位，因而也被称为布林带。其上下限范围不固定，随股价的滚动而变化。布林指标和麦克指标MIKE一样同属路径指标，股价波动在上限和下限的区间之内，这条带状区的宽窄，随着股价波动幅度的大小而变化，股价涨跌幅度加大时，带状区变宽，涨跌幅度狭小盘整时，带状区则变窄。
概念
布林线指标，即BOLL指标，其英文全称是“Bollinger Bands”，布林线(BOLL)由约翰 布林先生创造，其利用统计原理，求出股价的标准差及其信赖区间，从而确定股价的波动范围及未来走势，利用波带显示股价的安全高低价位，因而也被称为布林带。其上下限范围不固定，随股价的滚动而变化。布林指标和麦克指标MIKE一样同属路径指标，股价波动在上限和下限的区间之内，这条带状区的宽窄，随着股价波动幅度的大小而变化，股价涨跌幅度加大时，带状区变宽，涨跌幅度狭小盘整时，带状区则变窄。
算法
- 中轨线 = N日的移动平均线
- 上轨线 = 中轨线 + 两倍的标准差
- 下轨线 = 中轨线 – 两倍的标准差
通信达公式
BOLL: MA(CLOSE, M); # 收盘价的M日简单移动平均，一般m为20
UB: BOLL + 2 * STD(CLOSE, M); # BOLL + 2*收盘价的M日估算标准差
LB: BOLL - 2 * STD(CLOSE, M); # BOLL - 2*收盘价的M日估算标准差
STD(X,N) 返回估算标准差,即 平方根 (最近N日累计(收盘价 - MA) * (收盘价 - MA)/N)import pandas as pd
import numpy as np
# personal lib
from lib import lib
DataAPI.settings.cache_enabled = Falsedef SMA(x, n, m):
"""SMA(X,N,M):X的N日移动平均,M为权重,如Y=(X*M+Y'*(N-M))/N
e.g SMA(dataframe['close'], 10, 1)
return a list of sma
"""
x = list(x) if not isinstance(x, (list, tuple)) else x
sma = [0] * len(x) # 初始化为0，比如说10日均线，在上市前10天是没有数据的，应该设置为0或者是nan
for i in range(n-1, len(x)): # 从第 n 天开始计算移动平均，n之前都为0或nan
sma[i] = (x[i] * m + sma[i-1] * (n-m)) * 1.0 / n
return sma
def df_round(data, columns, decimals=2):
"""对 data 对几个 columns 取小数点精度
"""
for i in columns:
data[i] = data[i].round(decimals)
return data# @lib.wrapper_func_time
def cal_boll(ticker):
"""计算股票ticker 的 boll
"""
listDate = DataAPI.SecIDGet(assetClass='E', ticker=ticker, field=['listDate']).listDate[0]
data = DataAPI.MktEqudAdjGet(ticker=ticker, beginDate=listDate.replace('-',''), isOpen='1',
field=["secShortName", "tradeDate", "closePrice", "preClosePrice", "turnoverVol"])
data['ma_20'] = pd.rolling_mean(data.closePrice, 20)
tmp = [0] * len(data)
for i in range(19, len(data)):
tmp[i] = data.closePrice[ max(i-19, 0) : i + 1].std()
data['ma_std'] = tmp
data['midBoll'] = data['ma_20']
data['upperBoll'] = data['ma_20'] + 2 * data['ma_std']
data['lowerBoll'] = data['ma_20'] - 2 * data['ma_std']
data = df_round(data, ['ma_std', 'midBoll', 'upperBoll', 'lowerBoll'], 2)
return datadf = cal_boll('000002')
df.tail(5)作图
from lib import reportdf_tail = df.tail(100)# report library: https://uqer.datayes.com/community/share/56a5b221228e5b2047d916d9ax = df_tail.plot(x=['tradeDate'], y=['closePrice'], figsize=(20, 6))report.fig_style(ax, ['closePrice'])ax = df_tail.plot(x=['tradeDate'], y=['midBoll', 'upperBoll', 'lowerBoll'], kind='line', figsize=(20, 6), color=['r', 'g', 'b'])report.fig_style(ax, ['midBoll', 'upperBoll', 'lowerBoll'], legend_loc='best')8.RSI（Relative Strength Index），相对强弱指数，在1978年6月由Wells Wider提出，用来衡量证券自身内在相对强度,最早被应用于期货买卖，后来人们发现在众多的图表技术分析中，RSI理论和实践极其适合于股票市场的短线投资，于是被用于股票升跌的测量和分析中。它是通过特定时期内股价的变动情况来计算市场买卖力量对比，比较一段时期内的平均收盘涨数和平均收盘跌数来分析市场买沽盘的意向和实力，进而作出对未来市场的走势并判断股票价格内部本质强弱、推测价格未来变动方向的技术指标。
1. RSI原理
- RSI（Relative Strength Index），相对强弱指数，在1978年6月由Wells Wider提出，用来衡量证券自身内在相对强度,最早被应用于期货买卖，后来人们发现在众多的图表技术分析中，RSI理论和实践极其适合于股票市场的短线投资，于是被用于股票升跌的测量和分析中。它是通过特定时期内股价的变动情况来计算市场买卖力量对比，比较一段时期内的平均收盘涨数和平均收盘跌数来分析市场买沽盘的意向和实力，进而作出对未来市场的走势并判断股票价格内部本质强弱、推测价格未来变动方向的技术指标。
- 简单说，RSI是以数字的方法求买卖双方的力量对比。譬如100个人面对一件商品，如果50个人以上要买，竞相抬价，商品价格必涨。相反，如果50个人以上争着卖出，价格自然下跌。RSI理论认为，任何市价的大涨或大跌，均在0-100之间变动。根据常态分配，RSI值多在30-70之间变动，通常80甚至90时被认为市场已到达超买状态，市场价格会回落调整。当价格低跌至30以下即被认为是超卖状态，市价将出现反弹回升。
2. 计算公式
- RSI(N)＝[N日上升平均数÷(N日上升平均数＋N日下跌平均数)]×100
- N日上升平均数是在某段长为N日子里升幅数的平均，N日下跌平均数则是在同一段日子里跌幅数的平均。若计算9日RSI，需要找出前10日内的上升平均数及下跌平均数（各9个），如下是某股票每日收盘价的涨跌：
- 第一日23．70第二日27．90 \ + 4．20
第三日26．50 \ – 1．40
第四日29．60 \ + 3．10
第五日31．10 \ + 1．50
第六日29．40 \ – 1．70
第七日25．50 \ – 3．90
第八日28．90 \ + 3．40
第九日20．50 \ – 8．40
第十日23．20 \ + 2．80
──────────────────
9日内上涨平均值：(4.20+3.10+1.50+3.40+2.80)/9 = 15.0/9 = 1.67
9日内下跌平均值：(1.40+1.70+3.90+8.40)/9 = 15.4/9 = 1.71
第10日：上升平均数=(4.20+3.10+1.50+3.40+2.80)/9=1.67
：下降平均数=(1.40+1.70+3.90+8.40)/9=1.71
：RSI=[1.67÷(1.67+1.71)]×100=49.41
- 在行情软件中，RSI指数通常提供3条：N = [6，12，24]，一般对应于[ RSI1，RSI2，RSI3 ]。
- 文中我们使用UQER已经安装好的 TA-Lib 包计算RSI。
3. 买卖原则
- 超买超卖界定 Wells Wider推荐的默认时间跨度是14天，他论证了应用月周期28日的一半是有效的，在TA-Lib中RSI的默认计算周期为14。在实际的操作中，RSI指标具有滞后性，且其数值变动区间与市场形态和计算周期跨度。
- – 市场形态影响：在起伏不大的震荡市场中，通常规定70以上超买，30以下超卖；在变化比较剧烈的市场中，通常规定80以上超买，20以下超卖。
– 计算周期跨度：对于较短计算周期的12日RSI，通常规定80以上超买，20以下超卖；对于较长计算周期的24日RSI，通常规定70以上超买，30以下超卖。 - 买卖信号
- – 参考方案1：根据单RSI信号超买超卖区：超买区卖出，超卖区买入。
– 参考方案2：根据短期与长期RSI信号：
1）短期RSI值在30以下，由下向上交叉黄色的长期RSI值时为买入信号；
2）短期RSI值在70以上，由上向下交叉黄色的长期RSI值时为卖出信号。 - 通常RSI在40~60之间不太具有参考价值
4. RSI示例
- RSI数值示例
import talib as ta import pandas as pd####get the daily dataspdb = DataAPI.MktEqudAdjGet(secID='600000.XSHG',beginDate='20120101',endDate='20151229',field='tradeDate,secID,closePrice')####change the index of the dataspdb = spdb.set_index('tradeDate').sort_index()spdb.index = pd.to_datetime(spdb.index)####timeperiod = 14rsi = ta.RSI(spdb.closePrice.values)rsi1 = ta.RSI(spdb.closePrice.values,6)rsi2 = ta.RSI(spdb.closePrice.values,12)rsi3 = ta.RSI(spdb.closePrice.values,24)spdb['rsi'] = rsi ##timeperiod = 14spdb['rsi1'] = rsi1 ##timeperiod = 6spdb['rsi2'] = rsi2 ##timeperiod = 12spdb['rsi3'] = rsi3 ##timeperiod = 24spdb.tail()####figure displayspdb[['closePrice']].plot(figure = 'BenchMark',figsize=(12,4))spdb[['rsi']].plot(figsize=(12,4), kind='bar', xticks=[], color='b')spdb[['rsi1','rsi2','rsi3']].plot(figsize=(12,4))- RSI信号示例
- 参考方案1：根据RSI信号超买超卖区给出信号
buyThres = 30sellThres = 70rsiSig = []for index in spdb['rsi']: if index > sellThres: rsiSig.append(-1) elif index < buyThres: rsiSig.append(1) else: rsiSig.append(0)spdb['rsiSig'] = rsiSigspdb['rsiSig'].plot(figsize=(12,4))- RSI信号示例
- 参考方案2：根据短期与长期RSI信号
buyThres = 30sellThres = 70rsiSig = []spdb['rsiDif'] = spdb['rsi1']-spdb['rsi3']rsiSig = [0]*len(spdb['rsiDif'])for index in range(len(spdb['rsiDif'])): if index != 0 and index != len(spdb['rsiDif'])-1: if spdb['rsiDif'][index] > 0 and spdb['rsiDif'][index-1] < 0 and spdb['rsi'][index-1] < buyThres: rsiSig[index+1] = 1 elif spdb['rsiDif'][index] < 0 and spdb['rsiDif'][index-1] > 0 and spdb['rsi'][index-1] > sellThres: rsiSig[index+1] = -1spdb['rsiSig'] = rsiSigspdb['rsiSig'].plot(figsize=(12,4))5. RSI策略示例
在示例中，我们以两种信号生成方案分别对于招商银行个股和沪深300成分股进行了策略测试。- 信号生成方案1：根据RSI信号超买超卖区给出信号——招商银行个股
import talib as tadef initialize(account): account.buyThres = 30 account.sellThres = 70 def handle_data(account): closePrice = account.get_attribute_history('closePrice', history) rsi ={} for stock in account.universe: rsi[stock] = ta.RSI(closePrice[stock], history-1)[-1] for stock in account.universe: if (rsi[stock] < account.buyThres): order(stock, (round((1*account.cash/len(account.universe)/account.referencePrice[stock])/100)+1)*100) elif (rsi[stock] > account.sellThres): order_to(stock, 0) start = '2012-01-01'end = '2015-12-28'benchmark = 'HS300'universe = ['600000.XSHG']capital_base = 1000000history = 14- 信号生成方案1：根据RSI信号超买超卖区给出信号——沪深300成分股
import talib as taimport numpy as npdef initialize(account): account.buyThres = 30 account.sellThres = 70 def handle_data(account): closePrice = account.get_attribute_history('closePrice', history) rsi ={} for stock in account.universe: # print closePrice[stock] try: rsi[stock] = ta.RSI(closePrice[stock],history-1)[-1] except: rsi[stock] = 50 # if closePrice[stock].tolist().count('nan') == 0: # rsi[stock] = ta.RSI(closePrice[stock], history-1)[-1] # else: # rsi[stock] = 50 for stock in account.universe: if (rsi[stock] < account.buyThres): try: order(stock, (round((1*account.cash/len(account.universe)/account.referencePrice[stock])/100)+1)*100) except: pass elif (rsi[stock] > account.sellThres): order_to(stock, 0) start = '2012-01-01'end = '2015-12-28'benchmark = 'HS300'universe = set_universe('HS300')capital_base = 1000000history = 14
- 信号生成方案2：根据短期与长期RSI信号——招商银行个股
import talib as tadef initialize(account): account.buyThres = 30 account.sellThres = 70 def handle_data(account): closePrice = account.get_attribute_history('closePrice', history) rsi1 ={} rsi2 ={} rsiDif = {} for stock in account.universe: rsi1[stock] = ta.RSI(closePrice[stock],6)[-2:] rsi2[stock] = ta.RSI(closePrice[stock],24)[-2] rsiDif[stock] = rsi1[stock] - rsi2[stock] for stock in account.universe: if rsiDif[stock][0] < 0 and rsiDif[stock][1]>0 and rsi1[stock][0] < account.buyThres: order(stock, (round((1*account.cash/len(account.universe)/account.referencePrice[stock])/100)+1)*100) elif rsiDif[stock][0] > 0 and rsiDif[stock][1]<0 and rsi1[stock][0] > account.sellThres: order_to(stock, 0) start = '2012-01-01'end = '2015-12-28'benchmark = 'HS300'universe = ['600000.XSHG']# universe = ['600000.XSHG','000001.XSHE']capital_base = 1000000history = 30- 信号生成方案2：根据短期与长期RSI信号——沪深300成分股
import talib as taimport numpy as npdef initialize(account): account.buyThres = 30 account.sellThres = 70 def handle_data(account): closePrice = account.get_attribute_history('closePrice', history) rsi1 ={} rsi2 ={} rsiDif = {} for stock in account.universe: try: rsi1[stock] = ta.RSI(closePrice[stock],6)[-2:] rsi2[stock] = ta.RSI(closePrice[stock],24)[-2:] rsiDif[stock] = rsi1[stock] - rsi2[stock] except: rsi1[stock] = np.array([50,50]) rsi2[stock] = np.array([50,50]) rsiDif[stock] = rsi1[stock] - rsi2[stock] for stock in account.universe: if rsiDif[stock][0] < 0 and rsiDif[stock][1]>0 and rsi1[stock][0] < account.buyThres: try: order(stock, (round((1*account.cash/len(account.universe)/account.referencePrice[stock])/100)+1)*100) except: pass elif rsiDif[stock][0] > 0 and rsiDif[stock][1]<0 and rsi1[stock][0] > account.sellThres: order_to(stock, 0) start = '2012-01-01'end = '2015-12-28'benchmark = 'HS300'universe = set_universe('HS300')capital_base = 1000000history = 305. 指标评测
- RSI信号量较少，择时效果差强人意，入场时点较好；
- 利用短期和长期RSI线交叉择时比只根据RSI超买超卖区择时效果相对要好。
9.在传统的金融理论中，波动率是风险的度量。如果固定窗口长度，不断滚动计算这段窗口期内的指数收益率的波动率，则可以得到一系列的波动率时间序列。
使用单向波动率指标对指数进行择时
单向波动率差值指数择时
在传统的金融理论中，波动率是风险的度量。如果固定窗口长度，不断滚动计算这段窗口期内的指数收益率的波动率，则可以得到一系列的波动率时间序列。下面的代码是一个简单的实现：
import seaborn
import pandas as pd
import matplotlib.pyplot as plt
VOL_LEN = 10
df = DataAPI.MktIdxdGet(indexID='000300.ZICN', beginDate='20120101', endDate='20160801', field=['preCloseIndex', 'closeIndex'])
fig = pylab.figure(figsize=(12, 5))
df['closeIndex'].plot()df['ret'] = df['closeIndex'] / df['preCloseIndex']df['vol'] = pd.rolling_std(df['ret'], 10)fig = pylab.figure(figsize=(12, 5))df['vol'].plot()第一个图是指数的值，第二个图是短期波动率的值，对比两张图我们可以看出当波动率剧烈增加的时候，一般都是有剧烈的上涨或下跌的时候。简单来说波动率的升高意味着市场的涨跌正在变得越来越明显，更容易出现明确的单边行情。然而一个无法通过波动率看出的信息在于：到底是出现哪边的单边的行情呢？
为了从波动率中过滤出更多的信息，我们需要对波动率序列进行处理。我们这里来考虑最简单的一种处理：单向波动率。
这个概念很好理解，就是带符号的波动率，如果某天指数涨了，那么波动率就是正的，反之如果跌了就是负的。
MA_LEN = 40df['sym_vol'] = (df['ret'] > 1) * df['vol'] - (df['ret'] < 1) * df['vol']fig = pylab.figure(figsize=(12, 5))df['sym_vol'].plot()收益率被成功地加上了符号，但是这些数据波动实在太大，为了更清楚地看出单向波动率的情况，直接用最简单的MA方法来平滑之。
df['sym_vol_ma'] = pd.rolling_mean(df['sym_vol'], MA_LEN)
fig = pylab.figure(figsize=(12, 5))
df['sym_vol_ma'].plot()
嗯，这样一来就好看多了，从这个图上可以看出单向波动率是有“周期性”的，一段时间大于0，一段时间小于0。由于单向波动率是带符号的波动率，可以直接从中读出市场的涨跌情况，因此有一些很直观的猜测：
- 当单向波动率小于0时，意味着市场的“负向”波动率占主体，直到变为正的为止
- 当单向波动率大于0时，意味着市场的“正向”波动率占主体，直到变为负的为止
这个猜测可以直接转化成一个择时策略：单向波动率为正时持有指数，为负时空仓。
但显然直接使用这个论断来择时未免有点武断，因为在一般的震荡市场当中，单向波动率可能出现在0上下反复震荡的情况。为了避免这种情况，稍微再多想一想，波动率为负时持有指数是希望指数在目前下跌较多的情况下上涨，即希望出现反转；波动率为正时持有指数是希望指数在已经上涨的情况下保持上涨，即希望保持趋势。因此，可以直接把大盘分为“反转”、“趋势”、“震荡”三种情况，分别采取不同的择时思路。
那么接下来的问题是，如何把大盘分为“反转”、“趋势”、“震荡”三种情况呢？
答案可以很简单也可以很难。这里为了节省时间，直接采取最简单粗暴的方式：大盘已经跌了很多了，就归于反转；大盘已经涨了很多了，就归于趋势；其他情况都属于震荡。
尽管这样的分法很粗糙，可能会把不少“反转”和“趋势”归于“震荡”，但是由于策略是仅对指数进行择时的，在震荡的时候可以选择简单地持有指数，这样风险和收益与指数是一致的，所以不会有什么额外的损失。
基于这样的思路，可以很快地实现一个标的是指数ETF的择时策略。
from collections import deque
import numpy as np
IDXETF = '510300.XSHG' # 华泰柏瑞300
MA_LEN = 40 # 长期收益率及波动率平滑窗口
VOL_LEN = 10 # 波动率计算窗口
THRESHOLD = 0.05 # 简单大盘风格判断基准
start = '2013-08-01'
end = '2016-08-01'
benchmark = 'HS300'
universe = [IDXETF] # 只择时一只
capital_base = 1000000
freq = 'd'
refresh_rate = 1
def initialize(account):
account.vol_diff = deque(maxlen=MA_LEN) # 定长双端队列
def handle_data(account):
# 计算长期累计收益与短期波动率
if IDXETF in account.universe:
history = account.get_symbol_history(IDXETF, MA_LEN)
ret = history['closePrice'] / history['preClosePrice'] - 1.
cum_ret = history['closePrice'][-1] / history['closePrice'][0] - 1
new_std = ret[-VOL_LEN:].std()
else:
ret = [0]
cum_ret = new_std = 0
# 更新波动率差值
if ret[-1] > 0:
account.vol_diff.append(new_std)
elif ret[-1] < 0:
account.vol_diff.append(-new_std)
else:
account.vol_diff.append(0.)
if len(account.vol_diff) < MA_LEN:
observe('vol_diff', 0.)
observe('cum_ret', 0.)
else:
vol_diff_mean = sum(account.vol_diff) / MA_LEN
if cum_ret <= -THRESHOLD: # 反转，单向波动率差为负时满仓
if vol_diff_mean < 0:
order_pct_to(IDXETF, 1.)
else:
order_to(IDXETF, 0.)
elif cum_ret >= THRESHOLD: # 趋势，单向波动率差为正时满仓
if vol_diff_mean > 0:
order_pct_to(IDXETF, 1.)
else:
order_to(IDXETF, 0.)
else:
order_pct_to(IDXETF, 1)
observe('vol_diff', vol_diff_mean)
observe('cum_ret', cum_ret)但从结果来看，表现还是不错的，由于满仓时也只是100%持有指数，不可能出现上涨超过大盘的情况，所以所有的超额收益的来源都是避免下跌，所有超额损失的来源都是踏空上涨，就这两点而言，这个策略的表现还是不错的。
不过就整体的分析而言，每一步都有更多的点可以挖掘：
- 有其他的单向波动率的定义吗？
- 单向波动率的形态可以用来择时吗？
- 指数加权移动平均会怎么样？
- 有其他的判断大盘趋势的方法吗？
- 单向波动率可以和其他指标结合吗？
- ……
这些内容会在未来慢慢地加到这个以单向波动率为核心的分析框架里来。
10.顺势指标CCI由唐纳德拉姆伯特所创，是通过测量股价的波动是否已超出其正常范围，来预测股价变化趋势的技术分析指标。
一、CCI指标简介与构造
顺势指标CCI由唐纳德拉姆伯特所创，是通过测量股价的波动是否已超出其正常范围，来预测股价变化趋势的技术分析指标。计算方法参考《技术指标系列（五）——CCI的顺势而为》。
下面描绘出CCI与股价时序图走势
def cci(stock,start_date,end_date,windows): #设置股票，起始时间，以及CCI指标多少日
import pandas as pd
import numpy as np
from CAL.PyCAL import *
Alpha = 0.015
eq_TP = {}
eq_MATP = {}
eq_meanDev = {}
eq_CCI = {}
cal = Calendar('China.SSE')
windows = '-'+str(windows)+'B'
start_date = Date.strptime(start_date,"%Y%m%d")
end_date = Date.strptime(end_date,"%Y%m%d")
timeLength = cal.bizDatesList(start_date, end_date)
for i in xrange(len(timeLength)):
begin_date = cal.advanceDate(timeLength[i],windows,BizDayConvention.Unadjusted)
begin_date =begin_date.strftime("%Y%m%d")
timeLength[i] = timeLength[i].strftime("%Y%m%d")
eq_static = DataAPI.MktEqudAdjGet(secID=stock,beginDate=begin_date,endDate=timeLength[i],field=['secID','highestPrice','lowestPrice','closePrice'],pandas="1")
for stk in stock:
try:
eq_TP[stk] = np.array(eq_static[eq_static['secID'] == stk].mean(axis=1))
eq_MATP[stk] = sum(eq_TP[stk])/len(eq_TP[stk])
eq_meanDev[stk] = sum(abs(eq_TP[stk] - eq_MATP[stk]))/len(eq_TP[stk])
eq_CCI[stk].append((eq_TP[stk][-1] - eq_MATP[stk])/(Alpha * eq_meanDev[stk]))
except:
eq_CCI[stk] = []
Date = pd.DataFrame(timeLength)
eq_CCI = pd.DataFrame(eq_CCI)
cciSeries = pd.concat([Date,eq_CCI],axis =1)
cciSeries.columns = ['Date','CCI']
return cciSeriesdef cci_price_Plot(stock,start_date,end_date,windows):
cciSeries = cci(stock,start_date,end_date,windows)
closePrice = DataAPI.MktEqudAdjGet(secID=stock,beginDate=start_date,endDate=end_date,field=['closePrice'],pandas="1")
table = pd.merge(cciSeries,closePrice, left_index=True, right_index=True, how = 'inner')
return tableimport seaborn
import pandas as pd
import numpy as np
from CAL.PyCAL import *
cal = Calendar('China.SSE')
table = cci_price_Plot(['600000.XSHG'],'20080531','20150901',30) #绘制浦发银行的CCI与股价对比图
tableDate = table.set_index('Date')
tableDate.plot(figsize=(20,8),subplots = 1)
二、CCI指标简单应用
选取CCI处于100和150之间，开始处于上涨趋势的股票。关于windows，我们用quick_backtest做一个简单的优化
def cci(account,N=20):
Alpha = 0.015
eq_TP = {}
eq_MATP = {}
eq_meanDev = {}
eq_CCI = {}
eq_highPrice = account.get_attribute_history('highPrice',N)
eq_closePrice = account.get_attribute_history('closePrice',N)
eq_lowPrice = account.get_attribute_history('lowPrice',N)
for stk in account.universe:
eq_TP[stk] = (eq_highPrice[stk] + eq_closePrice[stk] + eq_lowPrice[stk])/3
eq_MATP[stk] = sum(eq_TP[stk])/len(eq_TP[stk])
eq_meanDev[stk] = sum(abs(eq_TP[stk] - eq_MATP[stk]))/len(eq_TP[stk])
eq_CCI[stk] = (eq_TP[stk][-1] - eq_MATP[stk])/(Alpha * eq_meanDev[stk])
return eq_CCIstart = '2010-08-01' # 回测起始时间
end = '2014-08-01' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = set_universe('HS300') # 证券池，支持股票和基金
capital_base = 100000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 20 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
sim_params = quartz.sim_condition.env.SimulationParameters(start, end, benchmark, universe, capital_base)
idxmap_all, data_all = quartz.sim_condition.data_generator.get_daily_data(sim_params)from CAL.PyCAL import *
import pandas as pd
import numpy as np
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
eq_CCI = cci(account,window)
buylist = []
for stk in account.universe:
try:
if eq_CCI[stk] > 100 and eq_CCI[stk] < 150:
buylist.append(stk)
except:
pass
for stk in account.valid_secpos:
order_to(stk, 0)
for stk in buylist[:]:
if stk not in account.universe or account.referencePrice[stk] == 0 or np.isnan(account.referencePrice[stk]):
bulist.remove(stk)
for stk in buylist:
order(stk, account.referencePortfolioValue/account.referencePrice[stk]/len(buylist))
print 'window annualized_return sharpe max_drawdown'
for window in range(10, 100, 5):
strategy = quartz.sim_condition.strategy.TradingStrategy(initialize, handle_data)
bt_test, acct = quartz.quick_backtest(sim_params, strategy, idxmap_all, data_all,refresh_rate = refresh_rate)
perf = quartz.perf_parse(bt_test, acct)
print ' {0:2d} {1:>7.4f} {2:>7.4f} {3:>7.4f}'.format(window, perf['annualized_return'], perf['sharpe'], perf['max_drawdown']) from CAL.PyCAL import *import pandas as pdimport numpy as npstart = '2010-08-01' # 回测起始时间end = '2014-08-01' # 回测结束时间benchmark = 'HS300' # 策略参考标准universe = set_universe('HS300') # 证券池，支持股票和基金capital_base = 100000 # 起始资金freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测refresh_rate = 20 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟def initialize(account): # 初始化虚拟账户状态 passdef handle_data(account): # 每个交易日的买入卖出指令 eq_CCI = cci(account,85) buylist = [] for stk in account.universe: try: if eq_CCI[stk] > 100 and eq_CCI[stk] < 150: buylist.append(stk) except: pass for stk in account.valid_secpos: order_to(stk, 0) for stk in buylist[:]: if stk not in account.universe or account.referencePrice[stk] == 0 or np.isnan(account.referencePrice[stk]): bulist.remove(stk) for stk in buylist: order(stk, account.referencePortfolioValue/account.referencePrice[stk]/len(buylist)) 样本外测试
from CAL.PyCAL import *
import pandas as pd
import numpy as np
start = '2014-08-01' # 回测起始时间
end = '2015-08-01' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = set_universe('HS300') # 证券池，支持股票和基金
capital_base = 100000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 20 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
eq_CCI = cci(account,85)
buylist = []
for stk in account.universe:
try:
if eq_CCI[stk] > 100 and eq_CCI[stk] < 150:
buylist.append(stk)
except:
pass
for stk in account.valid_secpos:
order_to(stk, 0)
for stk in buylist[:]:
if stk not in account.universe or account.referencePrice[stk] == 0 or np.isnan(account.referencePrice[stk]):
bulist.remove(stk)
for stk in buylist:
order(stk, account.referencePortfolioValue/account.referencePrice[stk]/len(buylist)) 11.佳庆离散指标（Chaikin Volatility，简称CVLT，VCI，CV）又称“佳庆变异率指数”，是通过测量一段时间内价格幅度平均值的变化来反映价格的离散程度。
import numpy as np
start = datetime(2011, 1, 1)
end = datetime(2015, 4, 27)
benchmark = 'HS300'
universe = set_universe('SH50')
capital_base = 100000
short_history = 30
longest_history = 60
pos_pieces = 10
enter_window = 20
exit_window = 10
N = 4
def initialize(account):
account.postion_size_hold = {}
for stk in universe:
account.postion_size_hold[stk] = 0
def handle_data(account):
print account.current_date
histCloseLong = account.get_attribute_history('closePrice', longest_history)
histLowLong = account.get_attribute_history('lowPrice', longest_history)
histHighLong = account.get_attribute_history('highPrice', longest_history)
histTurnoverLong = account.get_attribute_history('turnoverVol', longest_history)
histCloseShort = account.get_attribute_history('closePrice', short_history)
histLowShort = account.get_attribute_history('lowPrice', short_history)
histHighShort = account.get_attribute_history('highPrice', short_history)
histTurnoverShort = account.get_attribute_history('turnoverVol', short_history)
for stock in account.universe:
cnt_price = account.referencePrice[stock]
a1 = histCloseLong[stock] - histLowLong[stock]
b1 = histCloseLong[stock] - histHighLong[stock]
c1 = histHighLong[stock] - histLowLong[stock]
d1 = histTurnoverLong[stock]
adl = ((((a1)-(b1))/(c1)))*d1
a2 = histCloseShort[stock] - histLowShort[stock]
b2 = histCloseShort[stock] - histHighShort[stock]
c2 = histHighShort[stock] - histLowShort[stock]
d2 = histTurnoverShort[stock]
ads = ((((a2)-(b2))/(c2)))*d2
mean_cp1 = adl.mean()
mean_cp2 = ads.mean()
flag = mean_cp1 - mean_cp2
if flag > 0 and account.postion_size_hold[stock]<N:
order_to(stock, capital_base/pos_pieces/cnt_price/N)
account.postion_size_hold[stock] += 1
elif flag < 0 :
order_to(stock, 0)
account.postion_size_hold[stock] = 012.DMI指标又叫动向指标或趋向指标，是通过分析股票价格在涨跌过程中买卖双方力量均衡点的变化情况，即多空双方的力量的变化受价格波动的影响而发生由均衡到失衡的循环过程，从而提供对趋势判断依据的一种技术指标。其由美国技术分析大师威尔斯·威尔德（Wells Wilder）所创造，是一种中长期股市技术分析方法。
DMI指标又叫动向指标或趋向指标，是通过分析股票价格在涨跌过程中买卖双方力量均衡点的变化情况，即多空双方的力量的变化受价格波动的影响而发生由均衡到失衡的循环过程，从而提供对趋势判断依据的一种技术指标。其由美国技术分析大师威尔斯·威尔德（Wells Wilder）所创造，是一种中长期股市技术分析方法。
DMI指标体系的构建：
TR = SUM(MAX(MAX(HIGH – LOW, ABS(HIGH-REF(CLOSE,1))), ABS(LOW – REF(CLOSE, 1))), N)
HD = HIGH – REF(HIGH, 1)
LD = REF(LOW, 1) – LOW
DMP = SUM(IF(HD>0 AND HD>LD, HD, 0), N)
DMM = SUM(IF(LD>0 AND LD>HD, LD, 0), N)
PDI = DMP*100/TR
MDI = DMM*100/TR
DX = ABS(MDI – PDI)/(MDI + PDI)*100
ADX = MA(ABS(MDI – PDI)/(MDI + PDI)*100, M)
其中变量与函数定义如下：
CLOSE：引用收盘价(在盘中指最新价)
HIGH：引用最高价
LOW：引用最低价
REF(X, N)：引用X在N个周期前的值
ABS(X)：求X的绝对值
MAX(A, B)：求A，B中的较大者
SUM(X, N)：得到X在N周期内的总和
IF(C, A, B)：如果C成立返回A，否则返回B
此外，PDI简记为+DI，MDI简记为-DI；参数：N=14(默认)，M=14 (默认)。
实际上从数学上看，DX或ADX的构建并不一定需要PDI与MDI，有DMP和DMM就行了。计算PDI与MDI是将指标数值控制在0到100之间。
#计算某一天的股票DMP与DMM值
def eq_DMPandDMM(stk_list,current_date,N=14):
cal = Calendar('China.SSE')
period = '-' + str(N+1) + 'B'
begin_date = cal.advanceDate(current_date,period,BizDayConvention.Unadjusted)
end_date = cal.advanceDate(current_date,'-1B',BizDayConvention.Unadjusted)
eq_hd = {}
eq_ld = {}
dmp_sum = 0
dmm_sum = 0
eq_dmp = {}
eq_dmm = {}
eq_Price = DataAPI.MktEqudAdjGet(secID=stk_list,beginDate=begin_date.strftime('%Y%m%d'),endDate=end_date.strftime('%Y%m%d'),field=['secID','highestPrice','lowestPrice'],pandas="1")
avaiable_list = eq_Price['secID'].drop_duplicates().tolist()
eq_Price.set_index('secID',inplace=True)
for stk in avaiable_list:
if len(eq_Price.ix[stk]) == (N+1):
eq_hd[stk] = np.array(eq_Price.ix[stk]['highestPrice'][1:] - eq_Price.ix[stk]['highestPrice'][:-1])
eq_ld[stk] = np.array(eq_Price.ix[stk]['lowestPrice'][:-1] - eq_Price.ix[stk]['lowestPrice'][1:])
for i in xrange(len(eq_ld[stk])):
if eq_hd[stk][i] > 0 and eq_hd[stk][i] > eq_ld[stk][i]:
dmp_sum = dmp_sum + eq_hd[stk][i]
if eq_ld[stk][i] > 0 and eq_ld[stk][i] > eq_hd[stk][i]:
dmm_sum = dmm_sum + eq_ld[stk][i]
eq_dmp[stk] = dmp_sum
eq_dmm[stk] = dmm_sum
dmm_sum = 0
dmp_sum = 0
return eq_dmp,eq_dmm #计算某一天股票的TR值
def eq_TR(stk_list,current_date,N=14):
cal = Calendar('China.SSE')
period = '-' + str(N+1) + 'B'
begin_date = cal.advanceDate(current_date,period,BizDayConvention.Unadjusted)
end_date = cal.advanceDate(current_date,'-1B',BizDayConvention.Unadjusted)
eq_hl = {} #HIGH - LOW
eq_hc = {} #HIGH - CLOSE
eq_lc = {} #LOW - CLOSE
eq_tr = {}
tr_sum = 0
eq_Price = DataAPI.MktEqudAdjGet(secID=stk_list,beginDate=begin_date.strftime('%Y%m%d'),endDate=end_date.strftime('%Y%m%d'),field=['secID','highestPrice','lowestPrice','closePrice'],pandas="1")
avaiable_list = eq_Price['secID'].drop_duplicates().tolist()
eq_Price.set_index('secID',inplace=True)
for stk in avaiable_list:
if len(eq_Price.ix[stk]) == (N+1):
eq_hl[stk] = np.array(eq_Price.ix[stk]['highestPrice'][1:] - eq_Price.ix[stk]['lowestPrice'][1:])
eq_hc[stk] = np.array(eq_Price.ix[stk]['highestPrice'][1:] - eq_Price.ix[stk]['closePrice'][:-1])
eq_lc[stk] = np.array(eq_Price.ix[stk]['lowestPrice'][:-1] - eq_Price.ix[stk]['closePrice'][1:])
for i in xrange(len(eq_hl[stk])):
tr_sum = tr_sum + max(max(eq_hl[stk][i],abs(eq_hc[stk][i])),abs(eq_lc[stk][i]))
eq_tr[stk] = tr_sum
tr_sum = 0
return eq_tr #计算某一天股票的ADX
def eq_ADX(stk_list,current_date,N=14):
cal = Calendar('China.SSE')
period = '-' + str(N) + 'B'
begin_date = cal.advanceDate(current_date,period,BizDayConvention.Unadjusted)
end_date = cal.advanceDate(current_date,'-1B',BizDayConvention.Unadjusted)
timeSeries = cal.bizDatesList(begin_date,end_date)
eq_adx = {}
adx_sum = 0
#初始化eq_adx
eq_Price = DataAPI.MktEqudAdjGet(secID=stk_list,beginDate=begin_date.strftime('%Y%m%d'),endDate=end_date.strftime('%Y%m%d'),field=['secID','highestPrice','lowestPrice'],pandas="1")
avaiable_list = eq_Price['secID'].drop_duplicates().tolist()
eq_Price.set_index('secID',inplace=True)
for stk in avaiable_list:
if len(eq_Price.ix[stk]) == N:
eq_adx[stk] = 0
#计算ADX
for i in xrange(len(timeSeries)):
eq_dmp,eq_dmm = eq_DMPandDMM(stk_list,timeSeries[i],N)
for stk in eq_dmp:
if eq_dmp[stk] == 0 and eq_dmm[stk] == 0: #当DMP与DMM都为零时，认为无趋势DX=0
pass
else:
eq_adx[stk] = eq_adx[stk] + abs(eq_dmp[stk] - eq_dmm[stk])/(eq_dmp[stk] + eq_dmm[stk])*100
for stk in eq_adx:
eq_adx[stk] = eq_adx[stk] / len(timeSeries)
return eq_adx简单应用：
当DMP上穿DMM时，意味着，上涨倾向强于下跌倾向，一个买入信号生成。反之则反。而ADX用于反映趋向变动的程度，在买入信号时，ADX伴随上升，则预示股价的涨势可能更强劲。
import numpy as np
import pandas as pd
from CAL.PyCAL import *
start = '2012-08-01' # 回测起始时间
end = '2015-08-01' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = set_universe('HS300') # 证券池，支持股票和基金
capital_base = 1000000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 20 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
cal = Calendar('China.SSE')
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
eq_dmp_now,eq_dmm_now = eq_DMPandDMM(account.universe,account.current_date,14)
eq_adx = eq_ADX(account.universe,account.current_date,14)
yestoday = cal.advanceDate(account.current_date,'-1B',BizDayConvention.Unadjusted)
eq_dmp_before,eq_dmm_before = eq_DMPandDMM(account.universe,yestoday,14)
eq_adx_before = eq_ADX(account.universe,yestoday,14)
long_bucket = []
short_bucket = []
for stk in account.universe:
try:
if eq_dmp_now[stk] > eq_dmm_now[stk] and eq_dmp_before[stk] < eq_dmm_before[stk] and eq_adx[stk] > eq_adx_before[stk]:
long_bucket.append(stk)
else:
short_bucket.append(stk)
except:
pass
#调仓逻辑是调仓时将所有满足条件的股票等权
stk_num = len(account.valid_secpos) + len(long_bucket)
for stk in account.valid_secpos:
if stk in short_bucket:
order_to(stk,0)
stk_num = stk_num - 1
for stk in account.valid_secpos:
if stk not in short_bucket:
order_to(stk,account.referencePortfolioValue/account.referencePrice[stk]/stk_num)
for stk in long_bucket:
if stk not in account.avail_secpos:
order_to(stk,account.referencePortfolioValue/account.referencePrice[stk]/stk_num)13.TRIX(Triple Exponentially Smoothed Moving Average）中文名称：三重指数平滑移动平均，长线操作时采用本指标的讯号，可以过滤掉一些短期波动的干扰，避免交易次数过于频繁，造成部分无利润的买卖，及手续费的损失。
本指标是一项超长周期的指标，长时间按照本指标讯号交易，获利百分比大于损失百分比，利润相当可观。
前言
最近在看趋势择时，逛了下UQER，发现不少均线、MACD等指标的，但是木有看到关于TRIX的，就半参考着矿友accretion的策略Simple MACD做了个TRIX策略。
TRIX简介
TRIX(Triple Exponentially Smoothed Moving Average）中文名称：三重指数平滑移动平均，长线操作时采用本指标的讯号，可以过滤掉一些短期波动的干扰，避免交易次数过于频繁，造成部分无利润的买卖，及手续费的损失。
本指标是一项超长周期的指标，长时间按照本指标讯号交易，获利百分比大于损失百分比，利润相当可观。
计算公式
1.计算N日的指数移动平均线EMA
2.对上述EMA再进行两次N日指数移动平均后得到TR
3.TRIX = (TR – TR(昨日))/昨日TR * 100
4.MATRIX = TRIX的M日简单平均移动
TRIX的运用
1.当TRIX线一旦从下向上突破TRMA线，形成“金叉”时，预示着股价开始进入强势拉升阶段，投资者应及时买进股票。
2.当TRIX线向上突破TRMA线后，TRIX线和TRMA线同时向上运动时，预示着股价强势依旧，投资者应坚决持股待涨。
3.当TRIX线在高位有走平或掉头向下时，可能预示着股价强势特征即将结束，投资者应密切注意股价的走势，一旦K线图上的股价出现大跌迹象，投资者应及时卖出股票。
4.当TRIX线在高位向下突破TRMA线，形成“死叉”时，预示着股价强势上涨行情已经结束，投资者应坚决卖出余下股票，及时离场观望。
5.当TRIX线向下突破TRMA线后，TRIX线和TRMA线同时向下运动时，预示着股价弱势特征依旧，投资者应坚决持币观望。
6.当TRIX线在TRMA下方向下运动很长一段时间后，并且股价已经有较大的跌幅时，如果TRIX线在底部有走平或向上勾头迹象时，一旦股价在大的成交量的推动下向上攀升时，投资者可以及时少量地中线建仓。
7.当TRIX线再次向上突破TRMA线时，预示着股价将重拾升势，投资者可及时买入，持股待涨。
其中1、4为最重要的买入卖出信号，也是本策略的基石。
策略如下：
import pandas as pd
import numpy as np
start = '2011-08-01' # 回测起始时间
end = '2016-04-17' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = StockScreener(Factor.LCAP.nsmall(30)) # 因子选股，选取市值最小的30只股票作为备选
capital_base = 1000000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 5 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
N = 15 # 计算TR时的N
M = 45 # 计算MATRIX时的M
length_of_data = 3*N+M +10 # 取closeprice的天数，为了足够计算MATRIX、TRIX
all_close_prices = account.get_attribute_history('closePrice', length_of_data) # 获取历史closePrice数据
buy_list = [] # 备选买入清单
sell_list = [] # 卖出清单
for stk in account.universe:
prices = all_close_prices[stk]
if prices is None:
continue
try:
TRIX = talib.TRIX(prices,timeperiod=N) # 计算TRIX
MATRIX = talib.MA(TRIX,M,0) # 机选MATRIX
except:
continue
# 买入卖出判断
if (TRIX[-1]-MATRIX[-1]) > 0 and (TRIX[-5]-MATRIX[-5]) < 0: # 认为TRIX线向上突破TRMA线 金叉
buy_list.append(stk)
elif (TRIX[-1]-MATRIX[-1]) < 0 and (TRIX[-5]-MATRIX[-5]) > 0: # 认为TRIX线在高位向下突破TRMA线 死叉
sell_list.append(stk)
hold = []
buy = [] # 最终买入清单
# 买入卖出
for stk in account.valid_secpos:
# sell_list卖出
if stk in sell_list:
order_to(stk, 0)
# 其余继续持股
else:
hold.append(stk)
buy = hold
for stk in buy_list:
# 若buy_list中股票有未买入的，加入
if stk not in hold:
buy.append(stk)
if len(buy) > 0:
# 等仓位买入
amout = account.referencePortfolioValue/len(buy) # 每只股票买入数量
for stk in buy:
num = int(amout/account.referencePrice[stk] / 100.0) * 100
order_to(stk, num)
return
关于N、M的选取
N、M的选取将较大的影响策略效果，我也不是很懂如何去选，在TRIX的百度百科，最后有实践采用的是（24，72）的组合。
依葫芦画瓢，我以M=3N 试了下（9，27）、（24，72）、（15，45）的组合，其中（15，45）表现最优
不妨再试试其他的？下面再看看（12，9）和（12，72）
import pandas as pd
import numpy as np
start = '2011-08-01' # 回测起始时间
end = '2016-04-17' # 回测结束时间
benchmark = 'HS300' # 策略参考标准
universe = StockScreener(Factor.LCAP.nsmall(30)) # 因子选股，选取市值最小的30只股票作为备选
capital_base = 1000000 # 起始资金
freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测
refresh_rate = 5 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟
def initialize(account): # 初始化虚拟账户状态
pass
def handle_data(account): # 每个交易日的买入卖出指令
N = 12 # 计算TR时的N
M = 9 # 计算MATRIX时的M
length_of_data = 3*N+M +10 # 取closeprice的天数，为了足够计算MATRIX、TRIX
all_close_prices = account.get_attribute_history('closePrice', length_of_data) # 获取历史closePrice数据
buy_list = [] # 备选买入清单
sell_list = [] # 卖出清单
for stk in account.universe:
prices = all_close_prices[stk]
if prices is None:
continue
try:
TRIX = talib.TRIX(prices,timeperiod=N) # 计算TRIX
MATRIX = talib.MA(TRIX,M,0) # 机选MATRIX
except:
continue
# 买入卖出判断
if (TRIX[-1]-MATRIX[-1]) > 0 and (TRIX[-5]-MATRIX[-5]) < 0: # 认为TRIX线向上突破TRMA线 金叉
buy_list.append(stk)
elif (TRIX[-1]-MATRIX[-1]) < 0 and (TRIX[-5]-MATRIX[-5]) > 0: # 认为TRIX线在高位向下突破TRMA线 死叉
sell_list.append(stk)
hold = []
buy = [] # 最终买入清单
# 买入卖出
for stk in account.valid_secpos:
# sell_list卖出
if stk in sell_list:
order_to(stk, 0)
# 其余继续持股
else:
hold.append(stk)
buy = hold
for stk in buy_list:
# 若buy_list中股票有未买入的，加入
if stk not in hold:
buy.append(stk)
if len(buy) > 0:
# 等仓位买入
amout = account.referencePortfolioValue/len(buy) # 每只股票买入数量
for stk in buy:
num = int(amout/account.referencePrice[stk] / 100.0) * 100
order_to(stk, num)
return
import pandas as pdimport numpy as npstart = '2011-08-01' # 回测起始时间end = '2016-04-17' # 回测结束时间benchmark = 'HS300' # 策略参考标准universe = StockScreener(Factor.LCAP.nsmall(30)) # 因子选股，选取市值最小的30只股票作为备选capital_base = 1000000 # 起始资金freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测refresh_rate = 5 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟 def initialize(account): # 初始化虚拟账户状态 passdef handle_data(account): # 每个交易日的买入卖出指令 N = 12 # 计算TR时的N M = 72 # 计算MATRIX时的M length_of_data = 3*N+M +10 # 取closeprice的天数，为了足够计算MATRIX、TRIX all_close_prices = account.get_attribute_history('closePrice', length_of_data) # 获取历史closePrice数据 buy_list = [] # 备选买入清单 sell_list = [] # 卖出清单 for stk in account.universe: prices = all_close_prices[stk] if prices is None: continue try: TRIX = talib.TRIX(prices,timeperiod=N) # 计算TRIX MATRIX = talib.MA(TRIX,M,0) # 机选MATRIX except: continue # 买入卖出判断 if (TRIX[-1]-MATRIX[-1]) > 0 and (TRIX[-5]-MATRIX[-5]) < 0: # 认为TRIX线向上突破TRMA线 金叉 buy_list.append(stk) elif (TRIX[-1]-MATRIX[-1]) < 0 and (TRIX[-5]-MATRIX[-5]) > 0: # 认为TRIX线在高位向下突破TRMA线 死叉 sell_list.append(stk) hold = [] buy = [] # 最终买入清单 # 买入卖出 for stk in account.valid_secpos: # sell_list卖出 if stk in sell_list: order_to(stk, 0) # 其余继续持股 else: hold.append(stk) buy = hold for stk in buy_list: # 若buy_list中股票有未买入的，加入 if stk not in hold: buy.append(stk) if len(buy) > 0: # 等仓位买入 amout = account.referencePortfolioValue/len(buy) # 每只股票买入数量 for stk in buy: num = int(amout/account.referencePrice[stk] / 100.0) * 100 order_to(stk, num) return
小结
对比来看，似乎M越大，初期的效果越好。但是总体表现并无太大提高。
我这里是抛砖引玉、关于N,M的选择、策略性能的提高，希望大家讨论指点。
发布者：股市刺客，转载请注明出处：https://www.95sca.cn/archives/321273
站内所有文章皆来自网络转载或读者投稿，请勿用于商业用途。如有侵权、不妥之处，请联系站长并出示版权证明以便删除。敬请谅解！
