import pandas as pd
import numpy as np
import ta
import mplfinance as mpf
import matplotlib.pyplot as plt
from matplotlib.lines import Line2D
from matplotlib.patches import Rectangle
from matplotlib.collections import PolyCollection
def calculate_indicators(df):
# 强庄起爆绝底启动通达信副图指标计算
df['A1'] = df['high'].rolling(6).mean()
# 计算RSV
low_min = df['low'].rolling(13).min()
high_max = df['high'].rolling(13).max()
df['RSV'] = (df['close'] - low_min) / (high_max - low_min + 1e-8) * 100
df['RSV'] = df['RSV'].ewm(span=2, adjust=False).mean()
# 计算K、D、LS、GS、J
df['K'] = df['RSV'].ewm(span=2, adjust=False).mean()
df['D'] = df['K'].rolling(3).mean()
df['LS'] = df['K'] + df['D']
df['GS'] = df['LS'].shift(1).rolling(1).mean()
df['J'] = 3 * df['LS'] - 2 * df['GS']
# 计算财信号
df['A1J'] = abs(((3.48 * df['close'] + df['high'] + df['low']) / 4 -
df['close'].ewm(span=9, adjust=False).mean()) /
df['close'].ewm(span=9, adjust=False).mean())
df['A2J'] = (df['close'] * 2.15 + df['low'] + df['high']) / 4 * df['A1J']
df['X1'] = df[
'A2J'].ewm(span=4, adjust=False).mean()
df['X3'] = (df['close'].pct_change() * 100 > 8)
df['财'] = (df['close'] > df['X1']) & (df['X3'])
# 计算MACD双底启动
df['DIFF'] = df['close'].ewm(span=12, adjust=False).mean() - df['close'].ewm(span=26, adjust=False).mean()
df['DEA'] = df['DIFF'].ewm(span=8, adjust=False).mean()
df['MACD'] = 2 * (df['DIFF'] - df['DEA'])
df['A'] = np.where(df['DEA'] 0, 1, 0)
df['A1A'] = np.where((df['DIFF'] 0) & (df['DIFF'] > df['DEA']), 1, 0)
df['SS'] = np.where(df['A'] == 1, df['A1A'].cumsum(), 0)
df['SSB'] = df['SS'].expanding().max()
df['SSS'] = df['A1A'].cumsum()
df['双底启动'] = (df['SSS'] - df['SSB'] == 2) & (df['SSS'] > df['SSS'].shift(1))
# 计算绝底启动
n = 5
n1 = 21
df['LIJIN1'] = 4 * ((df['close'] - df['low'].rolling(n).min()) /
(df['high'].rolling(n).max() - df['low'].rolling(n).min() + 1e-8) * 100
df['LIJIN1'] = df['LIJIN1'].rolling(5).mean()
df['LIJIN1'] = 4 * df['LIJIN1'] - 3 * df['LIJIN1'].ewm(span=3, adjust=False).mean()
df['趋势线'] = np.where(df['LIJIN1'] > n1, df['LIJIN1'] - n1, 0)
df['绝底启动'
] = np.where(df['趋势线'].diff() > 0.01, 18, 0)
# 状态标记
df['持股'] = df['LS'] >= df['GS']
df['持币'] = df['LS'] <= df['GS']
return df
def plot_indicators(df):
# 创建强庄起爆绝底启动通达信副图指绘图区域
fig = plt.figure(figsize=(14, 10))
ax = plt.subplot()
# 绘制K线
mpf.plot(df, type='candle', ax=ax, show_nontrading=False, style='yahoo')
# 绘制警戒线
ax.axhline(160, color='green', linewidth=2, label='警戒线')
ax.axhline(100, color='white', linewidth=2, label='牛熊分界')
ax.axhline(20, color='magenta', linewidth=2, label='低吸起涨')
# 绘制LS和GS线
ax.plot(df.index, df['LS'], 'r-', label='LS')
ax.plot(df.index, df['GS'], 'y-', label='GS')
# 绘制带状区域
ax.fill_between(df.index, df['LS'], df['GS'],
where=df['LS']>=df['GS'],
facecolor='magenta', alpha=0.3)
ax.fill_between(df.index, df['LS'], df['GS'],
where=df['LS']<=df['GS'],
facecolor='blue', alpha=0.3)
# 绘制信号标记
# 高抛区
high_sell = df[df['J'] > 160]
for idx in high_sell.index:
ax.vlines(idx, 160, high_sell.loc[idx, 'J'],
colors='magenta', linewidths=1)
# 双底启动信号
double_bottom = df[df['双底启动']]
for idx in double_bottom.index:
ax.text(idx, 80, '---双底启动', color=
'magenta', fontsize=9,
horizontalalignment='center')
# 财神信号
fortune_god = df[df['财神']]
for idx in fortune_god.index:
ax.text(idx, 42, '加仓拉升', color='yellow', fontsize=9,
horizontalalignment='center')
# 设置强庄起爆绝底启动通达信副图指标图例和标题
ax.set_title('Technical Indicator Analysis', fontsize=16)
ax.set_ylabel('Price/Value', fontsize=12)
ax.legend(loc='upper left')
plt.tight_layout()
plt.show()
if __name__ == "__main__":
import yfinance as yf
data = yf.download('AAPL', start='2023-01-01', end='2023-12-31')
# 计算强庄起爆绝底启动通达信副图指标
df = calculate_indicators(data)
# 绘制图表
plot_indicators(df)
