import pandas as pd
from pyecharts import options as opts
from pyecharts.charts import Kline, Line, Bar, Grid, EffectScatter
from pyecharts.globals import SymbolType
from pyecharts.commons.utils import JsCode
def drawKline(df):
# kline
x = df.index.tolist()
y = list(df.loc[:, ['o', 'c', 'l', 'h']].round(2).values)
y = [i.tolist() for i in y] # 里面的单个数组也必须转换成list
# kline = Kline(init_opts=opts.InitOpts(width="100%", height="800px"), )
kline = Kline()
kline.add_xaxis(x)
kline.add_yaxis("kline", y)
kline.set_global_opts(
yaxis_opts=opts.AxisOpts(is_scale=True, splitline_opts=opts.SplitLineOpts(is_show=True)),
# tooltip_opts=opts.TooltipOpts(trigger="axis", axis_pointer_type="cross",is_show_content=False),
datazoom_opts=[
opts.DataZoomOpts(is_show=False, type_="inside", xaxis_index=[0, 0], range_end=50),
# xaxis_index=[0, 0]设置第一幅图为内部缩放
opts.DataZoomOpts(is_show=False, xaxis_index=[0, 1], pos_top="97%", range_end=50),
# xaxis_index=[0, 1]连接第二幅图的axis
],
)
# line
ma5 = df['c'].rolling(5).mean()
ma10 = df['c'].rolling(10).mean()
ma20 = df['c'].rolling(20).mean()
line = Line()
line.add_xaxis(x)
line.add_yaxis('MA5', ma5.round(2).tolist(), is_smooth=True)
line.add_yaxis('MA10', ma10.round(2).tolist(), is_smooth=True)
line.add_yaxis('MA20', ma20.round(2).tolist(), is_smooth=True)
line.set_series_opts(label_opts=opts.LabelOpts(is_show=False)) # 是否显示数据标签
# KD PD KK PK
es = EffectScatter()
# KD
v1 = df.index[df['KD'] == 1].tolist()
v2 = df['l'].loc[v1].tolist()
es.add_xaxis(v1)
es.add_yaxis('KD', v2, symbol=SymbolType.ARROW)
# PD
v1 = df.index[df['PD'] == 1].tolist()
v2 = df['h'].loc[v1].tolist()
es.add_xaxis(v1)
es.add_yaxis('PD', v2, symbol=SymbolType.ARROW, symbol_rotate=180)
# Kk
v1 = df.index[df['KK'] == 1].tolist()
v2 = (df['h'].loc[v1] * 1.01).tolist() # 向上偏移1% 防止重叠看不清
es.add_xaxis(v1)
es.add_yaxis('KK', v2, symbol=SymbolType.TRIANGLE, symbol_rotate=180)
# pk
v1 = df.index[df['PK'] == 1].tolist()
v2 = (df['l'].loc[v1] * 0.99).tolist() # 向下偏移1% 防止重叠看不清
es.add_xaxis(v1)
es.add_yaxis('PK', v2, symbol=SymbolType.TRIANGLE)
es.set_series_opts(label_opts=opts.LabelOpts(is_show=False)) # 是否显示数据标签
# overlap
# kline.overlap(line)
kline.overlap(es)
return kline
def drawMACD(df):
MACD = get_macd(df['c'], 12, 26, 9)
x = df.index.tolist()
bar_2 = Bar()
bar_2.add_xaxis(x)
bar_2.add_yaxis(
series_name="MACD",
y_axis=MACD.MACD.round(2).tolist(),
label_opts=opts.LabelOpts(is_show=False),
xaxis_index=1, # 用于合并显示时排列位置，单独显示不要添加
yaxis_index=1, # 用于合并显示时排列位置，单独显示不要添加
itemstyle_opts=opts.ItemStyleOpts(
color=JsCode(
"""
function(params) {
var colorList;
if (params.data >= 0) {
colorList = '#ef232a';
} else {
colorList = '#14b143';
}
return colorList;
}
"""
))
)
bar_2.set_global_opts(
xaxis_opts=opts.AxisOpts(
grid_index=1, # 用于合并显示时排列位置，单独显示不要添加
axislabel_opts=opts.LabelOpts(is_show=False),
),
yaxis_opts=opts.AxisOpts(
grid_index=1, # 用于合并显示时排列位置，单独显示不要添加
split_number=4,
),
legend_opts=opts.LegendOpts(is_show=False),
)
line_2 = Line()
line_2.add_xaxis(x)
line_2.add_yaxis(
series_name="DIF",
y_axis=MACD.DIF.round(2),
xaxis_index=1,
yaxis_index=2,
label_opts=opts.LabelOpts(is_show=False),
)
line_2.add_yaxis(
series_name="DEA",
y_axis=MACD.DEA.round(2),
xaxis_index=1,
yaxis_index=2,
label_opts=opts.LabelOpts(is_show=False),
)
line_2.set_global_opts(legend_opts=opts.LegendOpts(is_show=False))
return bar_2.overlap(line_2)
def drawAll(df):
df.index = df.index.strftime('%Y-%m-%d %H:%M:%S')
kline = drawKline(df)
MACD = drawMACD(df)
grid_chart = Grid(init_opts=opts.InitOpts(width="100%", height="800px"))
grid_chart.add(
kline,
grid_opts=opts.GridOpts(pos_left="3%", pos_right="1%", height="70%"),
)
grid_chart.add(
MACD,
grid_opts=opts.GridOpts(
pos_left="3%", pos_right="1%", pos_top="80%", height="20%"
)
)
grid_chart.render('123.html')
def get_macd(price, fast, slow, smooth):
exp1 = price.ewm(span=fast, adjust=False).mean()
exp2 = price.ewm(span=slow, adjust=False).mean()
DIF = exp1 - exp2
DEA = DIF.ewm(span=smooth, adjust=False).mean()
MACD = (DIF - DEA) * 2
columns1 = pd.DataFrame(DIF).rename(columns={'c': 'DIF'})
columns2 = pd.DataFrame(DEA).rename(columns={'c': 'DEA'})
columns3 = pd.DataFrame(MACD).rename(columns={'c': 'MACD'})
frames = [columns1, columns2, columns3]
df = pd.concat(frames, join='inner', axis=1)
return df