TATrend
趋势指标综合策略示例
本文档提供趋势指标的综合应用策略示例。
多均线策略 {#多均线策略}
使用多条均线判断趋势和交易时机。
python
def multi_ma_strategy():
"""多均线趋势策略"""
records = exchange.GetRecords()
if len(records) < 200:
return None
closes = [r['Close'] for r in records]
# 计算多条均线
sma20 = TA.SMA(closes, 20)
sma50 = TA.SMA(closes, 50)
sma200 = TA.SMA(closes, 200)
ema12 = TA.EMA(closes, 12)
ema26 = TA.EMA(closes, 26)
current_price = closes[-1]
# 判断趋势
if sma20[-1] > sma50[-1] > sma200[-1]:
trend = "强势多头"
elif sma20[-1] < sma50[-1] < sma200[-1]:
trend = "强势空头"
else:
trend = "震荡"
# EMA 金叉死叉
signal = None
if ema12[-1] > ema26[-1] and ema12[-2] <= ema26[-2]:
signal = "金叉"
elif ema12[-1] < ema26[-1] and ema12[-2] >= ema26[-2]:
signal = "死叉"
# 综合判断
if signal == "金叉" and trend == "强势多头":
return "强买入"
elif signal == "死叉" and trend == "强势空头":
return "强卖出"
elif signal == "金叉":
return "买入"
elif signal == "死叉":
return "卖出"
return None
def main():
while True:
action = multi_ma_strategy()
if action:
Log(f"交易信号: {action}")
Sleep(60000)均线支撑阻力 {#均线支撑阻力}
使用均线作为动态支撑阻力位。
python
def ma_support_resistance():
"""均线支撑阻力策略"""
records = exchange.GetRecords()
if len(records) < 100:
return None
closes = [r['Close'] for r in records]
current_price = closes[-1]
# 计算多条均线作为支撑阻力位
sma20 = TA.SMA(closes, 20)
sma50 = TA.SMA(closes, 50)
sma100 = TA.SMA(closes, 100)
sma200 = TA.SMA(closes, 200)
# 找出最近的支撑位（价格之下的均线）
supports = []
resistances = []
mas = {
'SMA20': sma20[-1],
'SMA50': sma50[-1],
'SMA100': sma100[-1],
'SMA200': sma200[-1]
}
for name, ma_val in mas.items():
if ma_val < current_price:
supports.append((name, ma_val))
elif ma_val > current_price:
resistances.append((name, ma_val))
result = {
'price': current_price,
'supports': sorted(supports, key=lambda x: x[1], reverse=True),
'resistances': sorted(resistances, key=lambda x: x[1])
}
# 判断交易机会
if supports:
nearest_support = supports[0]
distance = (current_price - nearest_support[1]) / nearest_support[1]
if distance < 0.01: # 接近支撑位
result['signal'] = f"接近{nearest_support[0]}支撑，考虑买入"
if resistances:
nearest_resistance = resistances[0]
distance = (nearest_resistance[1] - current_price) / current_price
if distance < 0.01: # 接近阻力位
result['signal'] = f"接近{nearest_resistance[0]}阻力，考虑卖出"
return result
def main():
while True:
result = ma_support_resistance()
if result:
Log(f"当前价格: {result['price']}")
Log(f"支撑位: {result.get('supports', [])}")
Log(f"阻力位: {result.get('resistances', [])}")
if 'signal' in result:
Log(result['signal'])
Sleep(60000)均线系统 {#均线系统}
完整的均线交易系统。
python
class MASystem:
def __init__(self):
self.position = None # 'long', 'short', None
self.entry_price = 0
def check_entry(self, closes):
"""检查入场信号"""
sma5 = TA.SMA(closes, 5)
sma10 = TA.SMA(closes, 10)
sma20 = TA.SMA(closes, 20)
sma50 = TA.SMA(closes, 50)
# 多头入场：短期均线上穿长期均线
if (sma5[-1] > sma10[-1] > sma20[-1] and
sma5[-2] <= sma10[-2] and
closes[-1] > sma50[-1]):
return 'long'
# 空头入场：短期均线下穿长期均线
if (sma5[-1] < sma10[-1] < sma20[-1] and
sma5[-2] >= sma10[-2] and
closes[-1] < sma50[-1]):
return 'short'
return None
def check_exit(self, closes):
"""检查出场信号"""
if not self.position:
return False
sma5 = TA.SMA(closes, 5)
sma10 = TA.SMA(closes, 10)
sma20 = TA.SMA(closes, 20)
current_price = closes[-1]
if self.position == 'long':
# 死叉或跌破SMA20
if (sma5[-1] < sma10[-1] and sma5[-2] >= sma10[-2]) or \
current_price < sma20[-1]:
return True
elif self.position == 'short':
# 金叉或突破SMA20
if (sma5[-1] > sma10[-1] and sma5[-2] <= sma10[-2]) or \
current_price > sma20[-1]:
return True
return False
def run(self):
"""运行策略"""
records = exchange.GetRecords()
if len(records) < 60:
return
closes = [r['Close'] for r in records]
current_price = closes[-1]
# 检查出场
if self.check_exit(closes):
Log(f"出场信号，持仓: {self.position}, 入场价: {self.entry_price}, 当前价: {current_price}")
self.position = None
self.entry_price = 0
return
# 检查入场
if not self.position:
entry_signal = self.check_entry(closes)
if entry_signal:
self.position = entry_signal
self.entry_price = current_price
Log(f"入场信号: {entry_signal}, 价格: {current_price}")
def main():
system = MASystem()
while True:
system.run()
Sleep(60000)EMA趋势跟踪
使用EMA的快速反应特性跟踪趋势。
python
def ema_trend_following():
"""EMA趋势跟踪策略"""
records = exchange.GetRecords()
if len(records) < 100:
return None
closes = [r['Close'] for r in records]
# 使用多条EMA
ema8 = TA.EMA(closes, 8)
ema21 = TA.EMA(closes, 21)
ema55 = TA.EMA(closes, 55)
current_price = closes[-1]
# 强势上涨
if ema8[-1] > ema21[-1] > ema55[-1]:
# 回踩EMA21买入
if closes[-2] < ema21[-2] and current_price >= ema21[-1]:
return {
'signal': '买入',
'reason': '上升趋势中回踩EMA21'
}
# 强势下跌
elif ema8[-1] < ema21[-1] < ema55[-1]:
# 反弹EMA21卖出
if closes[-2] > ema21[-2] and current_price <= ema21[-1]:
return {
'signal': '卖出',
'reason': '下降趋势中反弹EMA21'
}
# 金叉
if ema8[-1] > ema21[-1] and ema8[-2] <= ema21[-2]:
return {
'signal': '买入',
'reason': 'EMA8上穿EMA21金叉'
}
# 死叉
elif ema8[-1] < ema21[-1] and ema8[-2] >= ema21[-2]:
return {
'signal': '卖出',
'reason': 'EMA8下穿EMA21死叉'
}
return None
def main():
while True:
result = ema_trend_following()
if result:
Log(f"信号: {result['signal']}")
Log(f"原因: {result['reason']}")
Sleep(60000)