# 4.5 CCI ? CCI 順勢指標探索 > 來源：https://uqer.io/community/share/55f28179f9f06c1ea681fa40 ## 一、CCI指標簡介與構造 順勢指標CCI由唐納德拉姆伯特所創，是通過測量股價的波動是否已超出其正常范圍，來預測股價變化趨勢的技術分析指標。計算方法參考[《技術指標系列（五）——CCI的順勢而為 》](https://uqer.io/community/share/55f28179f9f06c1ea681fa40)。 下面描繪出CCI與股價時序圖走勢 ```py 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 cciSeries def 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 table ``` ```py 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) array([<matplotlib.axes.AxesSubplot object at 0x60037d0>, <matplotlib.axes.AxesSubplot object at 0x602fa90>], dtype=object) ```  ## 二、CCI指標簡單應用 選取CCI處于100和150之間，開始處于上漲趨勢的股票。關于windows，我們用quick_backtest做一個簡單的優化 ```py 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_CCI ``` ```py start = '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) ``` ```py 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']) window annualized_return sharpe max_drawdown 10 0.0186 -0.0610 0.4161 15 -0.0367 -0.2818 0.5448 20 0.0753 0.1734 0.4531 25 0.0268 -0.0254 0.3098 30 -0.0440 -0.3198 0.5640 35 0.0481 0.0599 0.4794 40 0.1117 0.3270 0.4057 45 0.0619 0.1176 0.2353 50 -0.0425 -0.3442 0.4226 55 0.0227 -0.0577 0.3355 60 0.0513 0.0540 0.4461 65 0.0860 0.1969 0.2304 70 0.0434 0.0218 0.3005 75 0.0126 -0.1176 0.3672 80 0.0891 0.2084 0.3728 85 0.1002 0.2554 0.2971 90 0.0768 0.1687 0.2710 95 0.0243 -0.0588 0.3461 ``` ```py from CAL.PyCAL import * import pandas as pd import numpy as np start = '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): # 初始化虛擬賬戶狀態 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)) ```  樣本外測試 ```py 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)) ```