指数CTA

CTA Hedge

数据预处理

from fxdayu_data.data import MongoHandler, StockData
from fxdayu_data.analysis.statistic import count_advances, frame_map
from datetime import datetime
import pandas as pd
import talib
import numpy as np
mh = MongoHandler(db='HS300')
sd = StockData(db='HS300')
sd.update_all()
def mapper(frame):
    close = pd.DataFrame(frame['close'].dropna(), dtype=np.float64)
    frame['ma'] = talib.abstract.MA(close, timeperiod=50)
    return frame
if __name__ == '__main__':
    candles = mh.read(mh.db.collection_names(), start=datetime(2007, 3, 1))
    candles = frame_map(candles, mapper)
    adv = count_advances(candles, start=50)
    mh.inplace(adv, 'Beta', 'HS_Index_Beta')

1. Advance & Decline Percent (Participation)
   

   $$Slope_{100}((Advance-Decline)/Total)$$

2. High & Low Percent (Leadership)
   

   $$Slope_{100}((High-Low)/Total)$$

3. MA50_up & MA50_down Percent (Trend)
   

   $$(MA50_{up}-MA50_{down})/Total$$

计算数据

from fxdayu_data.data import MongoHandler
from datetime import datetime
mh = MongoHandler(db='HS_Index_Beta')
beta = mh.read('Beta')
sh300 = mh.read('hs300.D', 'Global_index', start=datetime(2010, 03, 23))
beta['U-D/T'] = (beta['up']-beta['down'])/len(beta)*100
beta['H-L/T'] = (beta['high']-beta['low'])/len(beta)*100
beta['ma_u-d/T'] = (beta['ma_up']-beta['ma_down'])/len(beta)*100
mh.inplace(beta, 'Beta_Result', 'HS_Index_Beta')

处理显示

import talib as ta
import matplotlib.pyplot as plt
beta_r = mh.read('Beta_Result', 'HS_Index_Beta')
beta_r['slope_AD'] = ta.abstract.LINEARREG_SLOPE(beta_r, 100, price='U-D/T')
beta_r['slope_HL'] = ta.abstract.LINEARREG_SLOPE(beta_r, 100, price='H-L/T')
beta_r['SMA_S_AD'] = ta.abstract.MA(beta_r, 10, price='slope_AD')
beta_r['LMA_S_AD'] = ta.abstract.MA(beta_r, 40, price='slope_AD')
beta_r['SMA_S_HL'] = ta.abstract.MA(beta_r, 10, price='slope_HL')
beta_r['LMA_S_HL'] = ta.abstract.MA(beta_r, 40, price='slope_HL')
beta_r['MA_MA_UD'] = ta.abstract.MA(beta_r, 20, price='ma_u-d/T')
fig,(ax, ax1, ax2, ax3) = plt.subplots(4, 1, sharex=True)
ax.plot(sh300)
ax1.plot(beta_r.LMA_S_AD)
ax1.plot(beta_r.SMA_S_AD)
ax2.plot(beta_r.LMA_S_HL)
ax2.plot(beta_r.SMA_S_HL)
ax3.plot(beta_r['ma_u-d/T'])
ax3.plot(beta_r.MA_MA_UD)
ax.legend(loc='upper left')
ax1.legend(loc='upper left')
ax2.legend(loc='upper left')
ax3.legend(loc='upper left')
plt.show()

4.Volume

beta_r['hs300'] = hs300.close
beta_r['LMA_V'] = abstract.EMA(sh300, 20, price='volume')
beta_r['SMA_V'] = abstract.EMA(sh300, 40, price='volume')
plt.subplot(2,1,1)
plt.plot(beta_r.hs300)
plt.subplot(2,1,2)
plt.plot(beta_r.LMA_V)
plt.plot(beta_r.SMA_V)
plt.show()

5.用四条橘色线做Regression(AD/HL/MA/Volume)

beta_r = beta_r.dropna()
mlr = regression.linear_model.OLS(beta_r['hs300'], sm.add_constant(np.column_stack((beta_r['SMA_S_AD'],
                                                                                    beta_r['SMA_S_HL'],
                                                                                    beta_r['MA_MA_UD'],
                                                                                    beta_r['SMA_V'],
                                                                                    )))).fit()
prediction = mlr.params[0] + mlr.params[1]*beta_r['SMA_S_AD'] + mlr.params[2]*beta_r['SMA_S_HL']+ \
             mlr.params[3]*beta_r['MA_MA_UD'] + mlr.params[4]*beta_r['SMA_V']
prediction.name = 'Prediction'
print(mlr.params)
plt.subplot(2,1,1)
beta_r['hs300'].plot()
prediction.plot(color='y')
plt.xlabel('Price')
plt.legend()
plt.subplot(2,1,2)
bar = beta_r['hs300'] - prediction
plt.bar(bar.index, bar)
plt.hlines(500, bar.index[0], bar.index[-1], alpha=0.3)
plt.hlines(-500, bar.index[0], bar.index[-1], alpha=0.3)
plt.show()

6.A Pairs Percent

import numpy as np
import talib as ta
from fxdayu_data.data import MongoHandler
import pandas as pd
import matplotlib.pyplot as plt
mh = MongoHandler(db='Global_index')
x = np.array(range(1,21,1))
hs = mh.read('hs300.D')
def calculate(s):
    mas = ta.abstract.MA(hs, s)
    mal = ta.abstract.MA(hs, 4*s)
    df = pd.DataFrame({'long': mal, 'short': mas}).dropna()
    df['direction'] = map(lambda s, l: 1 if s > l else 0, df['short'], df['long'])
    return df['direction']
if __name__ == '__main__':
    df = pd.DataFrame({s: calculate(s) for s in x}).dropna()
    MA_sum =sum([s for key, s in df.iteritems()])*5
    plt.subplot(2,1,1)
    plt.plot(hs.close)
    plt.subplot(2,1,2)
    plt.plot(MA_sum, 'r')
    plt.hlines(60, MA_sum.index[0], MA_sum.index[-1])
    plt.hlines(40, MA_sum.index[0], MA_sum.index[-1])
    plt.show()