TAMomentum
动量指标综合策略示例
本文档提供动量指标的综合应用策略示例。
多指标趋势确认策略 {#多指标趋势确认}
使用多个动量指标确认趋势强度和方向。
python
def check_trend_strength():
"""使用多个动量指标确认趋势强度"""
records = exchange.GetRecords()
if len(records) < 50:
return None
highs = [r['High'] for r in records]
lows = [r['Low'] for r in records]
closes = [r['Close'] for r in records]
volumes = [r['Volume'] for r in records]
# 计算多个指标
adx = TA.ADX(highs, lows, closes, 14)
plus_di = TA.PLUS_DI(highs, lows, closes, 14)
minus_di = TA.MINUS_DI(highs, lows, closes, 14)
rsi = TA.RSI(closes, 14)
mfi = TA.MFI(highs, lows, closes, volumes, 14)
macd, signal, hist = TA.MACD(closes, 12, 26, 9)
# 强势上涨判断
if (adx[-1] > 25 and
plus_di[-1] > minus_di[-1] and
rsi[-1] > 50 and
mfi[-1] > 50 and
macd[-1] > signal[-1]):
return "强势上涨"
# 强势下跌判断
elif (adx[-1] > 25 and
minus_di[-1] > plus_di[-1] and
rsi[-1] < 50 and
mfi[-1] < 50 and
macd[-1] < signal[-1]):
return "强势下跌"
# 弱势震荡
elif adx[-1] < 20:
return "震荡整理"
return "趋势不明"
def main():
while True:
trend = check_trend_strength()
Log("当前趋势:", trend)
if trend == "强势上涨":
Log("多指标确认上涨，可持有多单")
elif trend == "强势下跌":
Log("多指标确认下跌，可持有空单")
elif trend == "震荡整理":
Log("震荡行情，使用区间策略")
Sleep(60000)超买超卖综合判断 {#超买超卖综合判断}
综合多个超买超卖指标，投票机制判断。
python
def check_overbought_oversold():
"""综合多个超买超卖指标"""
records = exchange.GetRecords()
if len(records) < 50:
return None
highs = [r['High'] for r in records]
lows = [r['Low'] for r in records]
closes = [r['Close'] for r in records]
volumes = [r['Volume'] for r in records]
# 计算指标
rsi = TA.RSI(closes, 14)
k, d = TA.STOCH(highs, lows, closes, 9, 3, 0, 3, 0)
cci = TA.CCI(highs, lows, closes, 20)
willr = TA.WILLR(highs, lows, closes, 14)
mfi = TA.MFI(highs, lows, closes, volumes, 14)
# 超买信号计数
overbought_count = 0
if rsi[-1] > 70: overbought_count += 1
if k[-1] > 80: overbought_count += 1
if cci[-1] > 100: overbought_count += 1
if willr[-1] > -20: overbought_count += 1
if mfi[-1] > 80: overbought_count += 1
# 超卖信号计数
oversold_count = 0
if rsi[-1] < 30: oversold_count += 1
if k[-1] < 20: oversold_count += 1
if cci[-1] < -100: oversold_count += 1
if willr[-1] < -80: oversold_count += 1
if mfi[-1] < 20: oversold_count += 1
# 返回结果和详细信息
result = {
'status': '正常',
'overbought_count': overbought_count,
'oversold_count': oversold_count,
'details': {
'RSI': rsi[-1],
'STOCH_K': k[-1],
'CCI': cci[-1],
'WILLR': willr[-1],
'MFI': mfi[-1]
}
}
if overbought_count >= 4:
result['status'] = '强烈超买'
elif overbought_count >= 3:
result['status'] = '超买'
elif oversold_count >= 4:
result['status'] = '强烈超卖'
elif oversold_count >= 3:
result['status'] = '超卖'
return result
def main():
while True:
result = check_overbought_oversold()
if result:
Log(f"超买超卖状态: {result['status']}")
Log(f"超买信号数: {result['overbought_count']}/5")
Log(f"超卖信号数: {result['oversold_count']}/5")
Log(f"详细数据: {result['details']}")
if result['status'] == "强烈超卖":
Log("4个或以上指标确认超卖，强买入机会")
elif result['status'] == "强烈超买":
Log("4个或以上指标确认超买，强卖出机会")
Sleep(60000)背离检测策略 {#背离检测}
系统化检测价格与指标的背离。
python
def detect_divergence(prices, indicator, period=10):
"""检测价格与指标的背离"""
if len(prices) < period or len(indicator) < period:
return None
# 顶背离：价格创新高，指标未创新高
if (prices[-1] > max(prices[-period:-1]) and
indicator[-1] < max(indicator[-period:-1])):
return "顶背离"
# 底背离：价格创新低，指标未创新低
if (prices[-1] < min(prices[-period:-1]) and
indicator[-1] > min(indicator[-period:-1])):
return "底背离"
return None
def comprehensive_divergence_check():
"""综合背离检测"""
records = exchange.GetRecords()
if len(records) < 30:
return None
closes = [r['Close'] for r in records]
highs = [r['High'] for r in records]
lows = [r['Low'] for r in records]
volumes = [r['Volume'] for r in records]
# 计算指标
rsi = TA.RSI(closes, 14)
macd, signal, hist = TA.MACD(closes, 12, 26, 9)
mfi = TA.MFI(highs, lows, closes, volumes, 14)
cci = TA.CCI(highs, lows, closes, 20)
# 检测各指标背离
divergences = {
'RSI': detect_divergence(closes, rsi, 10),
'MACD': detect_divergence(closes, macd, 10),
'MFI': detect_divergence(closes, mfi, 10),
'CCI': detect_divergence(closes, cci, 10)
}
# 统计背离信号
top_div_count = sum(1 for div in divergences.values() if div == "顶背离")
bottom_div_count = sum(1 for div in divergences.values() if div == "底背离")
result = {
'divergences': divergences,
'top_div_count': top_div_count,
'bottom_div_count': bottom_div_count
}
return result
def main():
while True:
result = comprehensive_divergence_check()
if result:
Log("背离检测结果:", result['divergences'])
if result['top_div_count'] >= 2:
Log(f"检测到{result['top_div_count']}个指标顶背离，价格可能回调")
if result['bottom_div_count'] >= 2:
Log(f"检测到{result['bottom_div_count']}个指标底背离，价格可能反弹")
Sleep(60000)RSI + MACD 双指标策略
经典的趋势 + 超买超卖组合。
python
def rsi_macd_strategy():
"""RSI + MACD 组合策略"""
records = exchange.GetRecords()
if len(records) < 50:
return None
closes = [r['Close'] for r in records]
# 计算指标
rsi = TA.RSI(closes, 14)
macd, signal, hist = TA.MACD(closes, 12, 26, 9)
# 强买入信号：RSI 超卖 + MACD 金叉
if rsi[-1] < 30 and macd[-1] > signal[-1] and macd[-2] <= signal[-2]:
return {
'signal': '强买入',
'reason': 'RSI超卖 + MACD金叉',
'rsi': rsi[-1],
'macd': macd[-1],
'signal': signal[-1]
}
# 强卖出信号：RSI 超买 + MACD 死叉
elif rsi[-1] > 70 and macd[-1] < signal[-1] and macd[-2] >= signal[-2]:
return {
'signal': '强卖出',
'reason': 'RSI超买 + MACD死叉',
'rsi': rsi[-1],
'macd': macd[-1],
'signal': signal[-1]
}
# 普通买入：MACD 金叉
elif macd[-1] > signal[-1] and macd[-2] <= signal[-2]:
return {
'signal': '买入',
'reason': 'MACD金叉',
'rsi': rsi[-1]
}
# 普通卖出：MACD 死叉
elif macd[-1] < signal[-1] and macd[-2] >= signal[-2]:
return {
'signal': '卖出',
'reason': 'MACD死叉',
'rsi': rsi[-1]
}
return None
def main():
while True:
signal = rsi_macd_strategy()
if signal:
Log(f"信号: {signal['signal']}")
Log(f"原因: {signal['reason']}")
Log(f"RSI: {signal.get('rsi', 'N/A')}")
Sleep(60000)多周期共振策略
不同时间周期的指标同时确认。
python
def multi_timeframe_resonance():
"""多周期共振策略"""
# 获取不同周期K线
records_1h = exchange.GetRecords(PERIOD_H1)
records_4h = exchange.GetRecords(PERIOD_H4)
records_1d = exchange.GetRecords(PERIOD_D1)
if len(records_1h) < 30 or len(records_4h) < 30 or len(records_1d) < 30:
return None
# 计算各周期RSI
rsi_1h = TA.RSI([r['Close'] for r in records_1h], 14)
rsi_4h = TA.RSI([r['Close'] for r in records_4h], 14)
rsi_1d = TA.RSI([r['Close'] for r in records_1d], 14)
# 计算各周期MACD
macd_1h, signal_1h, _ = TA.MACD([r['Close'] for r in records_1h], 12, 26, 9)
macd_4h, signal_4h, _ = TA.MACD([r['Close'] for r in records_4h], 12, 26, 9)
macd_1d, signal_1d, _ = TA.MACD([r['Close'] for r in records_1d], 12, 26, 9)
# 多周期RSI超卖共振
if rsi_1h[-1] < 30 and rsi_4h[-1] < 30 and rsi_1d[-1] < 30:
return {
'signal': '强买入',
'reason': '多周期RSI超卖共振',
'timeframes': '1H + 4H + 1D'
}
# 多周期RSI超买共振
elif rsi_1h[-1] > 70 and rsi_4h[-1] > 70 and rsi_1d[-1] > 70:
return {
'signal': '强卖出',
'reason': '多周期RSI超买共振',
'timeframes': '1H + 4H + 1D'
}
# 多周期MACD金叉共振
elif (macd_1h[-1] > signal_1h[-1] and
macd_4h[-1] > signal_4h[-1] and
macd_1d[-1] > signal_1d[-1]):
return {
'signal': '买入',
'reason': '多周期MACD金叉共振',
'timeframes': '1H + 4H + 1D'
}
return None
def main():
while True:
signal = multi_timeframe_resonance()
if signal:
Log(f"信号: {signal['signal']}")
Log(f"原因: {signal['reason']}")
Log(f"共振周期: {signal['timeframes']}")
Sleep(300000) # 5分钟检查一次