MACD
Moving Average Convergence/Divergence，意为异同移动平均线。它刻画的是股价变化的速度
MACD算法
| 指标 | 含义 | 公式 |
|---|---|---|
| 短期EMA | 短期收盘价指数移动均线(12天) | 前一日EMA(12)11/13 +今日收盘价2/13 |
| 长期EMA | 长期收盘价指数移动均线(26天) | 前一日EMA(26)25/27 +今日收盘价2/27 |
| DIF | 短期EMA与长期EMA差值 | EMA(12)-EMA(26) |
| DEA | DIF线的M日指数移动均钱 | 前一日DEA8/10 +今日DEA2/10 |
| MACD | DIF线与DEA线的差*2 | (DIF-DEA)*2 |
MACD实现
ewm
Pandas中指数加权移动窗口函数采用ewm函数+mean（）快捷计算MACD
bar
Matplotlib柱状图函数，高效绘制MACD中的柱状
class TestMACD(TestCase):
def cal_macd(self,df,fastperiod=12,slowperiod=26,signalperiod=9):
ewma12 = df['close'].ewm(span=fastperiod,adjust=False).mean()
ewma26 = df['close'].ewm(span=slowperiod,adjust=False).mean()
df['dif'] = ewma12 - ewma26
df['dea'] = df['dif'].ewm(span=signalperiod,adjust=False).mean()
df['bar'] = (df['dif'] - df['dea']) * 2
return df
def test_MACD(self):
file_name = "D:\lhjytest\demo.csv"
df = pd.read_csv(file_name)
df.columns = ["stock_id","date","close","open","high","low","volume"]
df = df[["date","close","open","high","low","volume"]]
df["date"] = pd.to_datetime(df["date"])
df_macd = self.cal_macd(df)
print(df_macd)
plt.figure()
df_macd['dea'].plot(color="red",label='dea')
df_macd['dif'].plot(color="blue",label='dif')
plt.legend(loc='best')
pos_bar = []
pos_index = []
neg_bar = []
neg_index = []
for index,row in df_macd.iterrows():
if(row['bar'] > 0) :
pos_bar.append(row['bar'])
pos_index.append(index)
else:
neg_bar.append(row['bar'])
neg_index.append(index)
# 大于0用红色表示
plt.bar(pos_index,pos_bar,width=0.5,color='red')
# 小于等于0则用绿色表示
plt.bar(neg_index,neg_bar,width=0.5,color='green')
major_index = df_macd.index[df_macd.index]
major_xtics = df_macd['date'][df_macd.index]
plt.xticks(major_index,major_xtics)
plt.setp(plt.gca().get_xticklabels(),rotation=30)
plt.grid(linestyle='-.')
plt.title('000001平安银行MACD图')
plt.rcParams['axes.unicode_minus'] = False
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.show()
KDJ
中文名叫随机指数。通过价格波动的真实波幅来反映价格走势的强弱和超买超卖现象，在价格尚未上升或下降之前发出买卖信号的一种技术分析指标,适用于短期行情走势分析
KDJ算法
| 指标 | 含义 | 公式 |
|---|---|---|
| RSV | 未成熟随机指标值 | (Cn-Ln)/(Hn-Ln)x100(N日) |
| K | 当天K值 | 2/3x前一日K值+1/3x当日RSV |
| D | 当天D值 | 2/3x前一日D值+1/3x当日K值 |
| J | 当天J值 | 3当日K值-2当日D值 |
| 备注 | 若无前一日K 值与D值，则可分别用50来代替 |
KDJ实现
rolling
Pandas中移动窗口函数
每个窗口都是指定的固定大小，快捷计算Ln与Hn
expanding
Pandas中扩展窗口函数
只设置最小的观测值数量，不固定窗口大小，实现累计计算，即不断扩展，连用expanding().max()->创新高
class TestKDJ(TestCase):
def cal_kdj(self,df):
low_list = df['low'].rolling(9,min_periods=9).min()
low_list.fillna(value=df['low'].expanding().min(),inplace=True)
high_list = df['high'].rolling(9,min_periods=9).max()
high_list.fillna(value=df['high'].expanding().max(),inplace=True)
rsv = (df['close'] - low_list) / (high_list - low_list) * 100
df['k'] = pd.DataFrame(rsv).ewm(com=2).mean()
df['d'] = df['k'].ewm(com=2).mean()
df['j'] = 3 * df['k'] - 2 * df['d']
return df
def test_KDJ(self):
file_name = "D:\lhjytest\demo.csv"
df = pd.read_csv(file_name)
df.columns = ["stock_id","date","close","open","high","low","volume"]
df = df[["date","close","open","high","low","volume"]]
df["date"] = pd.to_datetime(df["date"])
df_kdj = self.cal_kdj(df)
print(df_kdj)
plt.figure()
df_kdj['k'].plot(color="red",label='k')
df_kdj['d'].plot(color="yellow",label='d')
df_kdj['j'].plot(color="blue",label='j')
plt.legend(loc='best')
major_index = df_kdj.index[df_kdj.index]
major_xtics = df_kdj['date'][df_kdj.index]
plt.xticks(major_index,major_xtics)
plt.setp(plt.gca().get_xticklabels(),rotation=30)
