昨天成交了31.1亿,以为行情暂时告一段落,结果今天依然火爆,上涨了3.7%,昨天卖掉后考虑的就是等待下跌才去建仓买入了,所以今天也没有买入。
看了今天的行情后,内心又有一点小波澜,心里在盘算如果今天早上买了多少金额,就可以赚多少,不知道有没有同样心里的投资者朋友,如果有的话,建议冷静冷静,早上那个时候看不出来,还有就是一定要遵守自己的投资逻辑,坚持自己的方法,不要轻易去改变,很容易出问题,预计未来一段时间都不会进场,除非大跌!


相对强弱指数(Relative Strength Index, RSI) 是由 J. Welles Wilder 在1978年提出的经典动量振荡指标,用于衡量价格变动的速度和幅度,判断市场的超买或超卖状态。
RSI的计算基于一定周期(通常14天)内的平均涨幅和平均跌幅:
2. 计算平均涨幅(AU)和平均跌幅(AD):
3. 计算RS(相对强度)和RSI:
在通达信软件中,RSI 指标可以通过内置函数 RSI
或自定义公式实现:
RSI(CLOSE, N); // N为周期(默认14)
N:=14;
LC:=REF(CLOSE,1);
U:=MAX(CLOSE-LC,0); // 上涨幅度
D:=MAX(LC-CLOSE,0); // 下跌幅度
AU:=SMA(U,N,1); // Wilder平滑(等同于EMA)
AD:=SMA(D,N,1);
RS:=AU/AD;
RSI:100*(RS/(1+RS));
参数调整:
N
(如6、14、21)适应不同交易周期。使用 pandas
和 numpy
计算RSI:
import pandas as pd
def calculate_rsi(data, window=14):
delta = data['Close'].diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(window=window).mean()
avg_loss = loss.rolling(window=window).mean()
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
return rsi
# 示例
data = pd.read_csv('stock_data.csv')
data['RSI'] = calculate_rsi(data)
print(data.tail())
def calculate_rsi_wilder(data, window=14):
delta = data['Close'].diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
# Wilder平滑(近似EMA)
avg_gain = gain.ewm(alpha=1/window, adjust=False).mean()
avg_loss = loss.ewm(alpha=1/window, adjust=False).mean()
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
return rsi
data['RSI_Wilder'] = calculate_rsi_wilder(data)
import matplotlib.pyplot as plt
plt.figure(figsize=(12, 6))
plt.plot(data['Close'], label='Price')
plt.title('RSI Indicator')
plt.legend()
ax2 = plt.twinx()
ax2.plot(data['RSI'], color='orange', label='RSI (14)')
ax2.axhline(70, color='red', linestyle='--')
ax2.axhline(30, color='green', linestyle='--')
ax2.set_ylabel('RSI')
plt.show()
RSI 是经典且实用的技术指标,适用于股票、外汇、加密货币等多种市场。通过 通达信公式 和 Python代码 实现后,可进一步优化策略,提高交易胜率。建议回测验证后再用于实盘。
CCI(Commodity Channel Index,商品通道指数) 是由著名技术分析师Donald Lambert于1980年提出的一种技术分析工具。CCI原本用于商品市场的分析,但由于其对价格波动的高敏感性,后来被广泛应用于股市、期货、外汇等金融市场。CCI的核心作用是通过衡量当前价格与一段时间内的平均价格的偏离程度,来判断市场的超买和超卖状况,从而提供买卖信号。
CCI指标的计算基于一个周期内的典型价格(Typical Price,TP),然后通过这个典型价格与其移动平均值的差异来计算。具体的计算公式如下:
2. CCI公式:
CCI作为一个趋势跟踪指标,主要用于以下几个方面:
CCI在-100和+100之间波动,当CCI从负值区(例如-150以下)反转上升并突破-100时,可能表示市场开始出现上行趋势,适合寻找买入机会;反之,当CCI从正值区(例如+150以上)反转下降并突破+100时,可能表示市场开始出现下行趋势,适合寻找卖出机会。
CCI的周期一般使用14日,和RSI等指标一样,14日周期被广泛使用,因为它能有效捕捉到市场的短期波动。而根据具体市场的特性,也可以调整周期,适应不同的交易策略。较短的周期会增加CCI的波动性,适合短线交易;较长的周期则更能反映长期趋势。
在通达信中,CCI指标可以通过简单的公式来实现。具体的代码如下:
CCI: (CLOSE - MA(TYPICALPRICE, 14)) / (0.015 * MD(TYPICALPRICE, 14));
TYPICALPRICE: (HIGH + LOW + CLOSE) / 3;
MD: (ABS(TYPICALPRICE - MA(TYPICALPRICE, 14))) / 14;
上述公式中:
TYPICALPRICE
是计算出来的典型价格;MA(TYPICALPRICE, 14)
是典型价格的14日简单移动平均;MD(TYPICALPRICE, 14)
是典型价格与其14日移动平均的绝对偏差。利用Python的Pandas和Numpy库,我们可以轻松地计算CCI指标。下面是实现CCI的Python代码:
import pandas as pd
import numpy as np
def calculate_cci(data, n=14):
"""
计算CCI(顺势指标)
:param data: 包含数据的DataFrame,必须包含'High', 'Low', 'Close'列
:param n: 计算CCI的周期,默认为14
:return: 包含CCI值的Series
"""
# 计算典型价格
data['TP'] = (data['High'] + data['Low'] + data['Close']) / 3
# 计算典型价格的移动平均
data['MA_TP'] = data['TP'].rolling(window=n).mean()
# 计算典型价格与其移动平均的偏差
data['MD'] = data['TP'].rolling(window=n).apply(lambda x: np.mean(np.abs(x - np.mean(x))), raw=False)
# 计算CCI
data['CCI'] = (data['TP'] - data['MA_TP']) / (0.015 * data['MD'])
return data['CCI']
# 示例数据(包含'High', 'Low', 'Close'列)
data = {
'High': [130, 132, 133, 135, 136, 138, 140, 142, 144, 146, 148, 150, 152, 154, 156],
'Low': [128, 130, 131, 132, 134, 136, 137, 139, 141, 143, 145, 147, 149, 151, 153],
'Close': [129, 131, 132, 134, 135, 137, 139, 141, 143, 145, 147, 149, 151, 153, 155]
}
df = pd.DataFrame(data)
# 计算CCI指标
df['CCI'] = calculate_cci(df, n=14)
# 输出结果
print(df[['High', 'Low', 'Close', 'CCI']])
(High + Low + Close) / 3
,即每个交易日的三个价格(最高价、最低价和收盘价)取平均值。rolling(window=n)
方法计算典型价格的n日移动平均(MA_TP)。apply()
函数计算典型价格与其移动平均值之间的偏差(MD)。CCI = (TP - MA(TP)) / (0.015 * MD)
。在实际操作中,CCI可以通过以下几种方式应用:
CCI(顺势指标)是一个非常有效的市场趋势判断工具,通过计算当前价格与一定周期内的典型价格偏差,帮助投资者识别市场的超买、超卖情况和趋势反转信号。通过通达信和Python编程,我们能够轻松实现CCI指标,并应用于各种金融市场的技术分析中。结合其他技术指标,CCI可以为投资者提供更加全面的市场判断依据,帮助他们做出更为准确的投资决策。
BIAS(乖离率),也叫做离差率或偏离率,是金融市场中常用的技术分析指标之一。它衡量的是某个时间段内,当前价格与其移动平均线之间的偏离程度。通常,BIAS用来判断市场是否处于超买或超卖状态,进而为投资者提供买卖决策的参考。
BIAS的计算公式如下:
其中:
BIAS指标的计算基于收盘价与某个周期内的移动平均线之间的差异。假设我们选择一个n日的周期(例如10日),则BIAS的计算公式为:
其中,MA(n)
是n日的移动平均(通常是简单移动平均)。
在通达信中,我们可以通过以下公式来计算BIAS指标:
BIAS: (CLOSE - MA(CLOSE, 10)) / MA(CLOSE, 10) * 100;
在这段代码中:
CLOSE
是当前的收盘价;MA(CLOSE, 10)
是计算10日的简单移动平均线;BIAS
计算得到的是收盘价与10日均线之间的乖离率,以百分比表示。Python是一种功能强大的编程语言,结合Pandas和Numpy库,可以方便地计算并可视化BIAS指标。以下是使用Python计算BIAS指标的代码:
import pandas as pd
def calculate_bias(data, n=10):
"""
计算BIAS(乖离率)指标
:param data: 包含收盘价的DataFrame,要求有一列名为'close'
:param n: 移动平均的周期,默认为10
:return: 包含BIAS值的Series
"""
ma = data['close'].rolling(window=n).mean() # 计算n日移动平均
bias = (data['close'] - ma) / ma * 100 # 计算乖离率
return bias
# 示例数据
data = {
'close': [100, 102, 105, 108, 107, 110, 113, 115, 114, 116, 118, 120, 122, 125, 123, 126, 130, 128, 127, 129]
}
df = pd.DataFrame(data)
# 计算BIAS指标
df['BIAS'] = calculate_bias(df, n=10)
print(df)
calculate_bias
,该函数接收一个包含收盘价的DataFrame和一个周期n(默认为10)。rolling(window=n)
函数计算n日移动平均值。(CLOSE - MA) / MA * 100
计算BIAS值。BIAS(乖离率)是一个非常实用的市场分析工具,通过计算收盘价与移动平均线的偏离程度,帮助投资者判断市场的超买和超卖状态。它既可以用于短期交易决策,也可以帮助确认长期趋势的强弱。通过通达信平台或Python编程,我们可以轻松地实现BIAS指标,并应用于股票、期货等市场分析。结合其他技术指标,BIAS可以为投资者提供更全面、有效的市场洞察,助力投资决策。
BBI(Bull and Bear Index,多空指数)是金融市场中用于判断市场多空趋势的一种综合性指标。它通过计算不同周期的移动平均线来反映市场的多空力量对比,从而帮助投资者进行买卖决策。BBI指标通过综合短期和长期的市场信息,能够有效减少单一周期指标的滞后性,提供相对平滑和稳定的市场趋势判断。
BBI的计算公式如下:
BBI = (MA3 + MA6 + MA12 + MA24) / 4
其中,MA3、MA6、MA12、MA24分别是3日、6日、12日和24日的移动平均线。
通达信是一款广泛使用的股票分析软件,它允许用户自定义技术指标。下面是BBI指标在通达信中的实现代码:
BBI: (MA(CLOSE, 3) + MA(CLOSE, 6) + MA(CLOSE, 12) + MA(CLOSE, 24)) / 4;
这段代码定义了一个名为BBI的指标,BBI的值是3日、6日、12日和24日移动平均线的平均值。
Python是一种强大的编程语言,结合Pandas等数据分析库,可以方便地计算BBI指标。下面是使用Python计算BBI指标的具体实现:
import pandas as pd
import numpy as np
def calculate_bbi(data):
"""
计算BBI指标
:param data: 包含收盘价的DataFrame,要求有一列名为'close'
:return: 包含BBI值的Series
"""
ma3 = data['close'].rolling(window=3).mean()
ma6 = data['close'].rolling(window=6).mean()
ma12 = data['close'].rolling(window=12).mean()
ma24 = data['close'].rolling(window=24).mean()
bbi = (ma3 + ma6 + ma12 + ma24) / 4
return bbi
# 示例数据
data = {
'close': [100, 102, 104, 103, 105, 107, 109, 108, 110, 112, 111, 113, 115, 116, 118, 117, 119, 121, 120, 122]
}
df = pd.DataFrame(data)
# 计算BBI
df['BBI'] = calculate_bbi(df)
print(df)
在这个示例中,首先导入所需的Pandas库,然后定义了一个计算BBI的函数calculate_bbi
,该函数接收一个包含收盘价的DataFrame,并返回计算出的BBI值。最后,使用示例数据计算并打印BBI值。
BBI指标通过综合不同周期的移动平均线信息,能够有效地反映市场的多空趋势,减少单一周期指标的滞后性,为投资者提供更为可靠的买卖信号。在实际应用中,可以结合其他技术指标和市场信息,更全面地进行市场分析和投资决策。通达信和Python的实现方法为投资者提供了不同的工具和平台,方便进行BBI指标的计算和应用。
QMT是迅投公司提供的量化软件,是提供给各个券商,再由券商给各个投资者使用,有些券商申请门槛高,有些券商申请门槛低,今天我们先不考虑申请门槛的问题,我们先看看QMT到底怎么快速使用起来。第一步,找券商申请一个模拟安装包和账号,安装后的界面如下图,这里以国金的举例:
第二步,输入账号登录后的界面如下,注意这里选择行情和交易,不要选中 独立交易:
第三步,下载python库:
第四步,新建策略并获取数据,点击新建策略后,把之前的代码删除,只保留这些内容,如下图:
运行以后的结果如下:
第五步,获取数据,处理数据:
通过上面5个步骤就能获取到相应的数据,后期的话就是根据获取到数据内容进行编写逻辑,通过ContextInfo.get_market_data_ex这个函数获取的数据结构就是Dataframe格式,所以大家可以根据df的格式来处理,再到后面的话就是交易逻辑的判断及下单。
布林带(Bollinger Bands)由约翰·布林格(John Bollinger)在1980年代创立,是一种基于统计学原理的技术分析工具。它通过测量价格的标准差来构建动态的交易区间,能够有效反映价格波动范围和趋势强度,是识别超买超卖状态、趋势延续和反转信号的重要指标。
布林带由三条轨道线组成:
BOLL的计算基于统计学中的标准差概念:
MB = SMA(CLOSE, N)
StdDev = STD(CLOSE, N)
UB = MB + K × StdDev
LB = MB - K × StdDev
N:=20; K:=2;
MID:MA(CLOSE,N);
UPPER:MID+K*STD(CLOSE,N);
LOWER:MID-K*STD(CLOSE,N);
N:=20; K:=2;
MID:MA(CLOSE,N),COLORWHITE;
UPPER:MID+K*STD(CLOSE,N),COLORRED;
LOWER:MID-K*STD(CLOSE,N),COLORGREEN;
// 价格突破标记
UP_BREAK:=CROSS(CLOSE,UPPER);
DN_BREAK:=CROSS(LOWER,CLOSE);
DRAWICON(UP_BREAK,UPPER,1);
DRAWICON(DN_BREAK,LOWER,2);
// 轨道收窄标记
BANDWIDTH:=(UPPER-LOWER)/MID*100;
NARROW:=BANDWIDTH<REF(BANDWIDTH,1) AND BANDWIDTH<20;
WIDE:=BANDWIDTH>REF(BANDWIDTH,1) AND BANDWIDTH>40;
DRAWTEXT(NARROW,MID,'收窄'),COLORYELLOW;
DRAWTEXT(WIDE,MID,'扩张'),COLORBLUE;
// 趋势方向标记
TREND_UP:=MID>REF(MID,1) AND CLOSE>MID;
TREND_DN:=MID<REF(MID,1) AND CLOSE<MID;
FILLRGN(TREND_UP,MID,UPPER,COLORRED,COLORTRANSPARENT);
FILLRGN(TREND_DN,MID,LOWER,COLORGREEN,COLORTRANSPARENT);
// 百分比轨道
PERCENT_B:=(CLOSE-LOWER)/(UPPER-LOWER)*100;
DRAWTEXT_FIX(1,0.9,0,0,STRCAT('位置:',CON2STR(PERCENT_B,2))),COLORWHITE;
import pandas as pd
import numpy as np
def calculate_boll(close, n=20, k=2):
"""
计算布林带指标
:param close: 收盘价序列
:param n: 移动平均周期
:param k: 标准差乘数
:return: 中轨, 上轨, 下轨
"""
mid = close.rolling(n).mean()
std = close.rolling(n).std()
upper = mid + k * std
lower = mid - k * std
return mid, upper, lower
# 示例使用
# mid, upper, lower = calculate_boll(df['close'])
import talib
def talib_boll(close, n=20, k=2):
"""
使用TA-Lib计算布林带
"""
upper, mid, lower = talib.BBANDS(close,
timeperiod=n,
nbdevup=k,
nbdevdn=k,
matype=0) # 0=SMA
return mid, upper, lower
import matplotlib.pyplot as plt
from matplotlib import gridspec
def plot_boll(close, n=20, k=2):
mid, upper, lower = calculate_boll(close, n, k)
fig = plt.figure(figsize=(14, 10))
gs = gridspec.GridSpec(2, 1, height_ratios=[3, 1])
# 价格与布林带图表
ax1 = plt.subplot(gs[0])
ax1.plot(close.index, close, label='Close', color='black', linewidth=2)
ax1.plot(mid.index, mid, label=f'Middle Band ({n}SMA)', color='blue', alpha=0.7)
ax1.plot(upper.index, upper, label=f'Upper Band ({k}σ)', color='red', alpha=0.7)
ax1.plot(lower.index, lower, label=f'Lower Band ({k}σ)', color='green', alpha=0.7)
# 填充布林带区域
ax1.fill_between(upper.index, upper, lower, color='gray', alpha=0.1)
# 标记突破信号
up_break = (close > upper) & (close.shift(1) <= upper.shift(1))
dn_break = (close < lower) & (close.shift(1) >= lower.shift(1))
for date in close[up_break].index:
ax1.annotate('↑', xy=(date, close[date]), xytext=(0, 10),
textcoords='offset points', ha='center', color='red', fontsize=12)
for date in close[dn_break].index:
ax1.annotate('↓', xy=(date, close[date]), xytext=(0, -15),
textcoords='offset points', ha='center', color='green', fontsize=12)
ax1.set_title('Bollinger Bands', fontsize=14)
ax1.legend(loc='upper left')
ax1.grid(True, linestyle='--', alpha=0.5)
# 布林带宽图表
ax2 = plt.subplot(gs[1])
bandwidth = (upper - lower) / mid * 100
ax2.plot(bandwidth.index, bandwidth, label='Bandwidth %', color='purple')
ax2.axhline(20, color='orange', linestyle='--', alpha=0.7)
ax2.axhline(40, color='blue', linestyle='--', alpha=0.7)
ax2.set_title('Bollinger Bandwidth', fontsize=14)
ax2.legend(loc='upper left')
ax2.grid(True, linestyle='--', alpha=0.5)
plt.tight_layout()
plt.show()
# 示例使用
# plot_boll(df['close'])
def basic_boll_strategy(close, n=20, k=2):
mid, upper, lower = calculate_boll(close, n, k)
signals = pd.DataFrame(index=close.index)
signals['price'] = close
signals['mid'] = mid
signals['upper'] = upper
signals['lower'] = lower
# 生成交易信号
signals['signal'] = 0
signals.loc[(close > upper) & (close.shift(1) <= upper.shift(1)), 'signal'] = -1 # 上轨突破卖出
signals.loc[(close < lower) & (close.shift(1) >= lower.shift(1)), 'signal'] = 1 # 下轨突破买入
return signals
def enhanced_boll_strategy(close, n=20, k=2, trend_period=50):
mid, upper, lower = calculate_boll(close, n, k)
trend_ma = close.rolling(trend_period).mean()
signals = pd.DataFrame(index=close.index)
signals['price'] = close
signals['mid'] = mid
signals['upper'] = upper
signals['lower'] = lower
signals['trend'] = trend_ma
# 生成交易信号(增加趋势过滤)
signals['signal'] = 0
# 买入条件:价格突破下轨且处于上升趋势
buy_cond = (close < lower) & (close.shift(1) >= lower.shift(1)) & (close > trend_ma)
# 卖出条件:价格突破上轨且处于下降趋势
sell_cond = (close > upper) & (close.shift(1) <= upper.shift(1)) & (close < trend_ma)
signals.loc[buy_cond, 'signal'] = 1
signals.loc[sell_cond, 'signal'] = -1
return signals
def boll_rsi_strategy(close, rsi_period=14, rsi_upper=70, rsi_lower=30):
mid, upper, lower = calculate_boll(close)
# 计算RSI
delta = close.diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(rsi_period).mean()
avg_loss = loss.rolling(rsi_period).mean()
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
signals = pd.DataFrame(index=close.index)
signals['price'] = close
signals['mid'] = mid
signals['rsi'] = rsi
# 生成交易信号
signals['signal'] = 0
# 买入条件:价格接近下轨且RSI超卖
buy_cond = (close < mid) & (rsi < rsi_lower)
# 卖出条件:价格接近上轨且RSI超买
sell_cond = (close > mid) & (rsi > rsi_upper)
signals.loc[buy_cond, 'signal'] = 1
signals.loc[sell_cond, 'signal'] = -1
return signals
def boll_bandwidth_analysis(close, n=20, k=2):
mid, upper, lower = calculate_boll(close, n, k)
bandwidth = (upper - lower) / mid * 100
signals = pd.DataFrame(index=close.index)
signals['price'] = close
signals['bandwidth'] = bandwidth
# 识别极端收缩和扩张
signals['signal'] = 0
# 带宽收缩至历史低位,可能预示突破
signals.loc[bandwidth < bandwidth.rolling(50).quantile(0.1), 'signal'] = 1
# 带宽扩张至历史高位,可能预示趋势延续
signals.loc[bandwidth > bandwidth.rolling(50).quantile(0.9), 'signal'] = -1
return signals
def multi_timeframe_boll(daily_close, weekly_close):
# 计算日线BOLL
daily_mid, daily_upper, daily_lower = calculate_boll(daily_close)
# 计算周线BOLL
weekly_mid, weekly_upper, weekly_lower = calculate_boll(weekly_close)
# 对齐时间索引
weekly_mid = weekly_mid.reindex(daily_close.index, method='ffill')
weekly_upper = weekly_upper.reindex(daily_close.index, method='ffill')
weekly_lower = weekly_lower.reindex(daily_close.index, method='ffill')
signals = pd.DataFrame(index=daily_close.index)
signals['price'] = daily_close
signals['daily_mid'] = daily_mid
signals['weekly_mid'] = weekly_mid
# 生成交易信号
signals['signal'] = 0
# 买入条件:日线价格上穿周线中轨且周线趋势向上
buy_cond = (daily_close > weekly_mid) & (daily_close.shift(1) <= weekly_mid.shift(1)) & \
(weekly_mid > weekly_mid.shift(1))
# 卖出条件:日线价格下穿周线中轨且周线趋势向下
sell_cond = (daily_close < weekly_mid) & (daily_close.shift(1) >= weekly_mid.shift(1)) & \
(weekly_mid < weekly_mid.shift(1))
signals.loc[buy_cond, 'signal'] = 1
signals.loc[sell_cond, 'signal'] = -1
return signals
def boll_atr_strategy(high, low, close, atr_period=14):
# 计算BOLL
mid, upper, lower = calculate_boll(close)
# 计算ATR
tr = pd.DataFrame({
'h-l': high - low,
'h-pc': abs(high - close.shift(1)),
'l-pc': abs(low - close.shift(1))
}).max(axis=1)
atr = tr.rolling(atr_period).mean()
# 动态调整交易逻辑
signals = pd.DataFrame(index=close.index)
signals['price'] = close
signals['atr'] = atr
# 生成交易信号
signals['signal'] = 0
# 高波动率时使用轨道突破策略
high_vol = atr > atr.rolling(50).mean()
signals.loc[high_vol & (close > upper) & (close.shift(1) <= upper.shift(1)), 'signal'] = -1
signals.loc[high_vol & (close < lower) & (close.shift(1) >= lower.shift(1)), 'signal'] = 1
# 低波动率时使用中轨回归策略
low_vol = atr <= atr.rolling(50).mean()
signals.loc[low_vol & (close > mid) & (close.shift(1) <= mid.shift(1)), 'signal'] = -1
signals.loc[low_vol & (close < mid) & (close.shift(1) >= mid.shift(1)), 'signal'] = 1
return signals
import backtrader as bt
class BollStrategy(bt.Strategy):
params = (
('n', 20),
('k', 2),
('printlog', False),
)
def __init__(self):
# 计算布林带
self.mid = bt.indicators.SMA(self.data.close, period=self.p.n)
self.std = bt.indicators.StdDev(self.data.close, period=self.p.n)
self.upper = self.mid + self.p.k * self.std
self.lower = self.mid - self.p.k * self.std
# 交叉信号
self.price_cross_upper = bt.indicators.CrossOver(self.data.close, self.upper)
self.price_cross_lower = bt.indicators.CrossOver(self.data.close, self.lower)
def next(self):
if not self.position:
if self.price_cross_lower > 0: # 价格上穿下轨
self.buy()
elif self.price_cross_upper < 0: # 价格下穿上轨
self.close()
def log(self, txt, dt=None, doprint=False):
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()}, {txt}')
布林带指标作为一种融合了趋势跟踪和波动率分析的技术工具,为交易者提供了独特的价格行为分析视角。通过本文的系统介绍,读者应当掌握:
需要特别强调的是:
建议交易者先进行充分的模拟测试,再逐步投入实盘交易。将BOLL指标纳入完整的交易系统中,结合基本面分析、资金管理和交易心理,才能最大化其效用。
KDJ指标,又称随机指标(Stochastic Oscillator),由George Lane在1950年代提出,是技术分析领域最受欢迎的动量指标之一。它通过比较收盘价与一定周期内的价格范围,反映市场的超买超卖状态和潜在的趋势转折点。
KDJ指标由三条曲线构成:
KDJ的计算基于以下核心概念:
N:=9; M1:=3; M2:=3;
RSV:=(CLOSE-LLV(LOW,N))/(HHV(HIGH,N)-LLV(LOW,N))*100;
K:SMA(RSV,M1,1);
D:SMA(K,M2,1);
J:3*K-2*D;
N:=9; M1:=3; M2:=3;
RSV:=(CLOSE-LLV(LOW,N))/(HHV(HIGH,N)-LLV(LOW,N))*100;
K:SMA(RSV,M1,1),COLORWHITE;
D:SMA(K,M2,1),COLORYELLOW;
J:3*K-2*D,COLORMAGENTA;
// 超买超卖线
OVERBOUGHT:80,COLORGRAY;
OVERSOLD:20,COLORGRAY;
// 金叉死叉标记
JC:=CROSS(K,D);
SC:=CROSS(D,K);
DRAWICON(JC,D,1);
DRAWICON(SC,D,2);
// 顶底背离检测
LL:=LLV(L,20);
HH:=HHV(H,20);
LD:=BARSLAST(LL);
HD:=BARSLAST(HH);
BD:=REF(CLOSE,LD)>CLOSE AND REF(K,LD)<K AND LD>5;
TD:=REF(CLOSE,HD)<CLOSE AND REF(K,HD)>K AND HD>5;
DRAWTEXT(BD,D,'底背离'),COLORRED;
DRAWTEXT(TD,D,'顶背离'),COLORGREEN;
// 颜色填充
FILLRGN(K>D,K,D,COLORRED,COLORTRANSPARENT);
FILLRGN(D>K,K,D,COLORGREEN,COLORTRANSPARENT);
import pandas as pd
import numpy as np
def calculate_kdj(high, low, close, n=9, m1=3, m2=3):
"""
计算KDJ指标
:param high: 最高价序列
:param low: 最低价序列
:param close: 收盘价序列
:param n: RSV计算周期
:param m1: K值平滑周期
:param m2: D值平滑周期
:return: K, D, J 值
"""
min_low = low.rolling(n).min()
max_high = high.rolling(n).max()
rsv = (close - min_low) / (max_high - min_low) * 100
rsv = rsv.replace([np.inf, -np.inf], np.nan).fillna(50)
k = rsv.ewm(alpha=1/m1, adjust=False).mean()
d = k.ewm(alpha=1/m2, adjust=False).mean()
j = 3 * k - 2 * d
return k, d, j
# 示例使用
# k, d, j = calculate_kdj(df['high'], df['low'], df['close'])
import talib
def talib_kdj(high, low, close, fastk_period=9, slowk_period=3, slowd_period=3):
"""
使用TA-Lib计算KDJ指标
"""
slowk, slowd = talib.STOCH(high, low, close,
fastk_period=fastk_period,
slowk_period=slowk_period,
slowk_matype=0,
slowd_period=slowd_period,
slowd_matype=0)
slowj = 3 * slowk - 2 * slowd
return slowk, slowd, slowj
import matplotlib.pyplot as plt
from matplotlib import gridspec
def plot_kdj(high, low, close, n=9, m1=3, m2=3):
k, d, j = calculate_kdj(high, low, close, n, m1, m2)
fig = plt.figure(figsize=(14, 10))
gs = gridspec.GridSpec(2, 1, height_ratios=[2, 1])
# 价格图表
ax1 = plt.subplot(gs[0])
ax1.plot(close.index, close, label='Close', color='black', linewidth=2)
ax1.set_title('Price Chart', fontsize=14)
ax1.grid(True, linestyle='--', alpha=0.7)
ax1.legend(loc='upper left')
# KDJ图表
ax2 = plt.subplot(gs[1])
ax2.plot(k.index, k, label='K', color='blue', linewidth=1.5)
ax2.plot(d.index, d, label='D', color='orange', linewidth=1.5)
ax2.plot(j.index, j, label='J', color='purple', linewidth=1.5, alpha=0.7)
# 超买超卖线
ax2.axhline(80, color='red', linestyle='--', linewidth=0.8, alpha=0.7)
ax2.axhline(20, color='green', linestyle='--', linewidth=0.8, alpha=0.7)
ax2.fill_between(k.index, 80, 100, color='red', alpha=0.1)
ax2.fill_between(k.index, 0, 20, color='green', alpha=0.1)
# 金叉死叉标记
cross_up = (k > d) & (k.shift(1) <= d.shift(1))
cross_down = (k < d) & (k.shift(1) >= d.shift(1))
for date in k[cross_up].index:
ax2.annotate('↑', xy=(date, k[date]), xytext=(0, 10),
textcoords='offset points', ha='center', color='green', fontsize=12)
for date in k[cross_down].index:
ax2.annotate('↓', xy=(date, k[date]), xytext=(0, -15),
textcoords='offset points', ha='center', color='red', fontsize=12)
ax2.set_title('KDJ Indicator', fontsize=14)
ax2.legend(loc='upper left')
ax2.grid(True, linestyle='--', alpha=0.5)
plt.tight_layout()
plt.show()
# 示例使用
# plot_kdj(df['high'], df['low'], df['close'])
def basic_kdj_strategy(high, low, close):
k, d, j = calculate_kdj(high, low, close)
signals = pd.DataFrame(index=close.index)
signals['price'] = close
signals['k'] = k
signals['d'] = d
signals['j'] = j
# 生成交易信号
signals['signal'] = 0
signals.loc[(k > d) & (k.shift(1) <= d.shift(1)), 'signal'] = 1 # 金叉买入
signals.loc[(k < d) & (k.shift(1) >= d.shift(1)), 'signal'] = -1 # 死叉卖出
return signals
def enhanced_kdj_strategy(high, low, close, ma_period=30):
k, d, j = calculate_kdj(high, low, close)
ma = close.rolling(ma_period).mean()
signals = pd.DataFrame(index=close.index)
signals['price'] = close
signals['k'] = k
signals['d'] = d
signals['ma'] = ma
# 生成交易信号(增加过滤条件)
signals['signal'] = 0
# 买入条件:KDJ金叉且价格在均线上方且K从超卖区回升
buy_cond = (k > d) & (k.shift(1) <= d.shift(1)) & (close > ma) & (k.shift(1) < 30)
# 卖出条件:KDJ死叉且价格在均线下方且K从超买区回落
sell_cond = (k < d) & (k.shift(1) >= d.shift(1)) & (close < ma) & (k.shift(1) > 70)
signals.loc[buy_cond, 'signal'] = 1
signals.loc[sell_cond, 'signal'] = -1
return signals
def kdj_boll_strategy(high, low, close, n=20, dev=2):
k, d, j = calculate_kdj(high, low, close)
ma = close.rolling(n).mean()
std = close.rolling(n).std()
upper = ma + dev * std
lower = ma - dev * std
signals = pd.DataFrame(index=close.index)
signals['price'] = close
signals['k'] = k
signals['d'] = d
signals['upper'] = upper
signals['lower'] = lower
# 生成交易信号
signals['signal'] = 0
# 买入条件:KDJ金叉且价格接近布林带下轨
buy_cond = (k > d) & (k.shift(1) <= d.shift(1)) & (close < lower + 0.2*(upper-lower))
# 卖出条件:KDJ死叉且价格接近布林带上轨
sell_cond = (k < d) & (k.shift(1) >= d.shift(1)) & (close > upper - 0.2*(upper-lower))
signals.loc[buy_cond, 'signal'] = 1
signals.loc[sell_cond, 'signal'] = -1
return signals
def multi_timeframe_kdj(daily_high, daily_low, daily_close,
weekly_high, weekly_low, weekly_close):
# 计算日线KDJ
daily_k, daily_d, _ = calculate_kdj(daily_high, daily_low, daily_close)
# 计算周线KDJ
weekly_k, weekly_d, _ = calculate_kdj(weekly_high, weekly_low, weekly_close)
# 对齐时间索引
weekly_k = weekly_k.reindex(daily_close.index, method='ffill')
weekly_d = weekly_d.reindex(daily_close.index, method='ffill')
signals = pd.DataFrame(index=daily_close.index)
signals['price'] = daily_close
signals['daily_k'] = daily_k
signals['daily_d'] = daily_d
signals['weekly_k'] = weekly_k
signals['weekly_d'] = weekly_d
# 生成交易信号(日线和周线同向时交易)
signals['signal'] = 0
# 买入条件:日线金叉且周线KDJ向上
buy_cond = (daily_k > daily_d) & (daily_k.shift(1) <= daily_d.shift(1)) & \
(weekly_k > weekly_k.shift(1))
# 卖出条件:日线死叉且周线KDJ向下
sell_cond = (daily_k < daily_d) & (daily_k.shift(1) >= daily_d.shift(1)) & \
(weekly_k < weekly_k.shift(1))
signals.loc[buy_cond, 'signal'] = 1
signals.loc[sell_cond, 'signal'] = -1
return signals
def adaptive_kdj_strategy(high, low, close, atr_period=14):
# 计算ATR衡量市场波动性
tr = pd.DataFrame({
'h-l': high - low,
'h-pc': abs(high - close.shift(1)),
'l-pc': abs(low - close.shift(1))
}).max(axis=1)
atr = tr.rolling(atr_period).mean()
# 根据波动率动态调整KDJ参数
volatility_level = atr / close
n = np.where(volatility_level > 0.02, 7,
np.where(volatility_level > 0.01, 9, 14))
# 计算动态KDJ
k, d, j = calculate_kdj(high, low, close, n=n)
signals = pd.DataFrame(index=close.index)
signals['price'] = close
signals['k'] = k
signals['d'] = d
signals['n'] = n
# 生成交易信号
signals['signal'] = 0
signals.loc[(k > d) & (k.shift(1) <= d.shift(1)), 'signal'] = 1
signals.loc[(k < d) & (k.shift(1) >= d.shift(1)), 'signal'] = -1
return signals
import backtrader as bt
class KdjStrategy(bt.Strategy):
params = (
('n', 9),
('m1', 3),
('m2', 3),
('printlog', False),
)
def __init__(self):
# 计算KDJ指标
self.high = self.data.high
self.low = self.data.low
self.close = self.data.close
# 计算RSV
self.llv = bt.indicators.Lowest(self.low, period=self.p.n)
self.hhv = bt.indicators.Highest(self.high, period=self.p.n)
self.rsv = (self.close - self.llv) / (self.hhv - self.llv) * 100
# 计算K,D
self.k = bt.indicators.EMA(self.rsv, period=self.p.m1)
self.d = bt.indicators.EMA(self.k, period=self.p.m2)
# 交叉信号
self.crossover = bt.indicators.CrossOver(self.k, self.d)
def next(self):
if not self.position:
if self.crossover > 0 and self.k[0] < 30: # 金叉且从超卖区回升
self.buy()
elif self.crossover < 0 and self.k[0] > 70: # 死叉且从超买区回落
self.close()
def log(self, txt, dt=None, doprint=False):
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()}, {txt}')
KDJ指标作为经典的技术分析工具,以其对价格波动的敏感性和明确的超买超卖信号,成为众多交易者不可或缺的分析手段。然而,任何技术指标都不是万能的,KDJ同样需要在正确的市场环境中配合其他分析工具使用。
通过本文从原理到实现的全面介绍,读者应当理解:
技术分析的本质是概率游戏,KDJ提供的是一种提高胜率的工具。持续学习、实践验证、纪律执行,才是技术分析发挥效用的根本保证。建议交易者结合自身风险偏好和市场特点,开发个性化的KDJ交易系统,并通过严格的回测和模拟交易验证后再投入实盘。
MACD(Moving Average Convergence Divergence,指数平滑异同移动平均线)是由Gerald Appel于1970年代提出的经典技术分析指标。它通过计算不同周期的指数移动平均线(EMA)之间的差异,来判断市场趋势的强度和方向,是股票、期货、外汇等金融市场中最受欢迎的技术分析工具之一。
MACD指标由三部分组成:
MACD的计算基于指数移动平均线(EMA),主要步骤如下:
DIF:EMA(CLOSE,12)-EMA(CLOSE,26);
DEA:EMA(DIF,9);
MACD:(DIF-DEA)*2,COLORSTICK;
SHORT:=12;
LONG:=26;
MID:=9;
DIF:EMA(CLOSE,SHORT)-EMA(CLOSE,LONG);
DEA:EMA(DIF,MID);
MACD:(DIF-DEA)*2,COLORSTICK;
// 添加零轴参考线
Zero:0,COLORGRAY;
// 金叉死叉标记
JC:=CROSS(DIF,DEA);
SC:=CROSS(DEA,DIF);
DRAWICON(JC,DEA,1);
DRAWICON(SC,DEA,2);
// 柱状图颜色设置
STICKLINE(MACD>0,0,MACD,0.8,0),COLORRED;
STICKLINE(MACD<0,0,MACD,0.8,0),COLORGREEN;
// 顶底背离判断
LL:=LLV(L,60);
HH:=HHV(H,60);
LD:=BARSLAST(LL);
HD:=BARSLAST(HH);
BD:=REF(CLOSE,LD)>CLOSE AND REF(DIF,LD)<DIF AND LD>10;
TD:=REF(CLOSE,HD)<CLOSE AND REF(DIF,HD)>DIF AND HD>10;
DRAWTEXT(BD,DEA,'底背离'),COLORRED;
DRAWTEXT(TD,DEA,'顶背离'),COLORGREEN;
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
def calculate_ema(prices, period):
return prices.ewm(span=period, adjust=False).mean()
def calculate_macd(close_prices, short=12, long=26, signal=9):
ema_short = calculate_ema(close_prices, short)
ema_long = calculate_ema(close_prices, long)
dif = ema_short - ema_long
dea = calculate_ema(dif, signal)
macd = (dif - dea) * 2
return dif, dea, macd
# 示例使用
# 假设df是一个包含收盘价的DataFrame
# dif, dea, macd = calculate_macd(df['close'])
import talib
def talib_macd(close_prices, short=12, long=26, signal=9):
dif, dea, macd = talib.MACD(close_prices,
fastperiod=short,
slowperiod=long,
signalperiod=signal)
return dif, dea, macd
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from matplotlib import gridspec
def plot_macd(close_prices, short=12, long=26, signal=9):
# 计算MACD
ema_short = close_prices.ewm(span=short, adjust=False).mean()
ema_long = close_prices.ewm(span=long, adjust=False).mean()
dif = ema_short - ema_long
dea = dif.ewm(span=signal, adjust=False).mean()
macd = (dif - dea) * 2
# 创建图表
fig = plt.figure(figsize=(12, 8))
gs = gridspec.GridSpec(2, 1, height_ratios=[3, 1])
# 价格图表
ax1 = plt.subplot(gs[0])
ax1.plot(close_prices.index, close_prices, label='Close Price', color='black')
ax1.plot(ema_short.index, ema_short, label=f'EMA {short}', color='blue', alpha=0.5)
ax1.plot(ema_long.index, ema_long, label=f'EMA {long}', color='red', alpha=0.5)
ax1.set_title('Price with EMA Lines')
ax1.legend()
# MACD图表
ax2 = plt.subplot(gs[1])
ax2.plot(dif.index, dif, label='DIF', color='blue')
ax2.plot(dea.index, dea, label='DEA', color='orange')
ax2.bar(macd.index, macd, label='MACD', color=np.where(macd > 0, 'red', 'green'))
ax2.axhline(0, color='gray', linestyle='--')
ax2.set_title('MACD Indicator')
ax2.legend()
plt.tight_layout()
plt.show()
# 示例使用
# 假设有一个包含收盘价的Series
# plot_macd(df['close'])
def macd_trading_strategy(close_prices):
dif, dea, macd = calculate_macd(close_prices)
signals = pd.DataFrame(index=close_prices.index)
signals['price'] = close_prices
signals['dif'] = dif
signals['dea'] = dea
signals['macd'] = macd
# 生成交易信号
signals['signal'] = 0
signals['signal'][dif > dea] = 1 # 买入信号
signals['signal'][dif < dea] = -1 # 卖出信号
# 计算持仓变化
signals['positions'] = signals['signal'].diff()
return signals
def enhanced_macd_strategy(close_prices, ma_period=50):
dif, dea, macd = calculate_macd(close_prices)
ma = close_prices.rolling(ma_period).mean()
signals = pd.DataFrame(index=close_prices.index)
signals['price'] = close_prices
signals['dif'] = dif
signals['dea'] = dea
signals['ma'] = ma
# 生成交易信号(增加过滤条件)
signals['signal'] = 0
# 买入条件:DIF上穿DEA且价格在均线上方
signals.loc[(dif > dea) & (dif.shift(1) <= dea.shift(1)) & (close_prices > ma), 'signal'] = 1
# 卖出条件:DIF下穿DEA且价格在均线下方
signals.loc[(dif < dea) & (dif.shift(1) >= dea.shift(1)) & (close_prices < ma), 'signal'] = -1
# 计算持仓变化
signals['positions'] = signals['signal'].diff()
return signals
def macd_rsi_strategy(close_prices, rsi_period=14, rsi_upper=70, rsi_lower=30):
# 计算MACD
dif, dea, macd = calculate_macd(close_prices)
# 计算RSI
delta = close_prices.diff()
gain = delta.where(delta > 0, 0)
loss = -delta.where(delta < 0, 0)
avg_gain = gain.rolling(rsi_period).mean()
avg_loss = loss.rolling(rsi_period).mean()
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
signals = pd.DataFrame(index=close_prices.index)
signals['price'] = close_prices
signals['dif'] = dif
signals['dea'] = dea
signals['rsi'] = rsi
# 生成交易信号
signals['signal'] = 0
# 买入条件:MACD金叉且RSI从超卖区回升
signals.loc[(dif > dea) & (dif.shift(1) <= dea.shift(1)) &
(rsi > rsi_lower) & (rsi.shift(1) <= rsi_lower), 'signal'] = 1
# 卖出条件:MACD死叉且RSI从超买区回落
signals.loc[(dif < dea) & (dif.shift(1) >= dea.shift(1)) &
(rsi < rsi_upper) & (rsi.shift(1) >= rsi_upper), 'signal'] = -1
# 计算持仓变化
signals['positions'] = signals['signal'].diff()
return signals
def multi_timeframe_macd(daily_close, weekly_close):
# 计算日线MACD
daily_dif, daily_dea, _ = calculate_macd(daily_close)
# 计算周线MACD
weekly_dif, weekly_dea, _ = calculate_macd(weekly_close)
# 对齐时间索引
weekly_dif = weekly_dif.reindex(daily_close.index, method='ffill')
weekly_dea = weekly_dea.reindex(daily_close.index, method='ffill')
signals = pd.DataFrame(index=daily_close.index)
signals['price'] = daily_close
signals['daily_dif'] = daily_dif
signals['daily_dea'] = daily_dea
signals['weekly_dif'] = weekly_dif
signals['weekly_dea'] = weekly_dea
# 生成交易信号(日线和周线同向时交易)
signals['signal'] = 0
# 买入条件:日线金叉且周线MACD在零轴上方
signals.loc[(daily_dif > daily_dea) & (daily_dif.shift(1) <= daily_dea.shift(1)) &
(weekly_dif > 0) & (weekly_dea > 0), 'signal'] = 1
# 卖出条件:日线死叉且周线MACD在零轴下方
signals.loc[(daily_dif < daily_dea) & (daily_dif.shift(1) >= daily_dea.shift(1)) &
(weekly_dif < 0) & (weekly_dea < 0), 'signal'] = -1
return signals
import backtrader as bt
class MacdStrategy(bt.Strategy):
params = (
('short', 12),
('long', 26),
('signal', 9),
('printlog', False),
)
def __init__(self):
self.macd = bt.indicators.MACD(self.data.close,
period_me1=self.p.short,
period_me2=self.p.long,
period_signal=self.p.signal)
self.crossover = bt.indicators.CrossOver(self.macd.macd, self.macd.signal)
def next(self):
if not self.position:
if self.crossover > 0: # MACD线上穿信号线
self.buy()
elif self.crossover < 0: # MACD线下穿信号线
self.close()
def log(self, txt, dt=None, doprint=False):
if self.params.printlog or doprint:
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()}, {txt}')
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status == order.Completed:
if order.isbuy():
self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}')
else:
self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}')
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
def notify_trade(self, trade):
if not trade.isclosed:
return
self.log(f'OPERATION PROFIT, GROSS {trade.pnl:.2f}, NET {trade.pnlcomm:.2f}')
# 使用示例
# cerebro = bt.Cerebro()
# data = bt.feeds.YahooFinanceData(dataname='AAPL', fromdate=datetime(2020,1,1), todate=datetime(2023,1,1))
# cerebro.adddata(data)
# cerebro.addstrategy(MacdStrategy)
# cerebro.run()
# cerebro.plot()
MACD作为经典技术分析工具,历经数十年市场检验依然被广泛使用,其核心价值在于简洁有效地捕捉趋势变化。然而,没有任何指标是完美的,MACD同样需要在正确的市场环境中配合其他分析工具使用。随着计算技术的发展,传统MACD正在与量化分析、机器学习等新方法结合,展现出新的生命力。
无论是通过通达信等传统交易软件,还是使用Python等编程语言实现,理解MACD的核心原理和适用条件都是关键。投资者应当根据自身交易风格和市场特点,灵活调整MACD参数和使用方法,将其纳入完整的交易体系中,而非孤立依赖。
技术分析的本质是概率游戏,MACD提供的是一种提高胜率的工具,而非确定性预测。持续学习、实践验证、纪律执行,才是技术分析发挥效用的根本保证。