用 AI 过滤交易信号
我在基于规则的逻辑之上添加了一个小的 Gemini AI 过滤器。机器人计算设置，构建最近市场状况的快照，并询问 Gemini 是否环境看起来足够干净还是太混乱。
一键发币： x402兼容 | Aptos | X Layer | SUI | SOL | BNB | ETH | BASE | ARB | OP | Polygon | Avalanche
我想要一个小型的机器人来帮我监控 EURUSD，遵循一套明确的规则，并且只在市场看起来良好的时候发送清晰的警报。我并不是试图预测未来或建立一个神奇的赚钱机器。我只是想要结构。这个想法很简单：让机器人做枯燥的部分，遵守规则，并避免过度交易。
大多数基础机器人一旦有任何指标交叉就会触发。这可能是一个单一的 EMA 交叉或一个简单的 RSI 水平。在实际交易中，你从不依赖于一个因素。你会看趋势、动量、波动性、蜡烛大小，甚至一天中的时间。我希望我的机器人更像一个耐心的交易者。它应该等待多个条件达成一致，然后才发送信号。
为了使它更有趣，我在基于规则的逻辑之上添加了一个小的 Gemini AI 过滤器。机器人计算设置，构建最近市场状况的快照，并询问 Gemini 是否环境看起来足够干净还是太混乱。Gemini 只能说 APPROVE 或 SKIP。如果它批准，信号就会通过。如果它跳过，机器人会忽略该设置并继续前进。
1、工作原理
这个机器人是用 Python 编写的，并在一个简单的循环中运行。它使用 yfinance 获取 EURUSD 蜡烛数据，使用 pandas 和 numpy 处理数据，使用 pytz 处理时区，使用 requests 与 Telegram 通信。如果存在 Gemini API 密钥，脚本还会导入 google.generativeai 库并启用 AI 过滤器。如果密钥缺失或未安装该库，机器人将不带 AI 层运行，并退回到纯基于规则的系统。
在脚本的顶部，我保存了所有主要的设置。机器人在 1 分钟的时间框架上监视 EURUSD 符号。我定义 EMA 快速和慢速周期为 9 和 21，RSI 周期为 14，MACD 快速、慢速和信号周期，以及布林带参数。我还设置了检查间隔为 60 秒，因此主循环每分钟运行一次。
另一个重要的参数是 ONE_SIGNAL_PER_HOUR。当此标志打开时，机器人将在每个小时开始时尝试创建新信号。这意味着它仍然每分钟唤醒一次，但除非当前分钟是零，否则会忽略详细的信号逻辑。这保持了警报的稀有性和易于跟踪。
2、将机器人连接到 Telegram
我希望机器人感觉像一个通知流，而不是一个复杂的终端程序。因此，所有输出都发送到 Telegram 聊天。在脚本中，一个小的帮助函数构建对 Telegram Bot API 的请求，发送消息，并打印响应状态。
每当机器人启动时，它会发送一条简短的消息，说明高级 EURUSD 信号机器人已上线，并且仅用于演示和教育用途。稍后，当出现新信号或交易结果准备就绪时，机器人会发送一条清晰的消息，包含方向、价格、时间和信号原因。每天结束时，它还会发送一个简短的每日总结，包括盈利和亏损。
3、从 Yahoo Finance 获取 EURUSD 数据
每次机器人想检查信号时，它首先从 Yahoo Finance 获取新鲜的 EURUSD 数据。它下载两天的一分钟蜡烛，以便所有指标都有足够的历史数据进行计算。然后函数清理数据并确保有一个适当的 Close 列。Yahoo Finance 有时会返回多级列，所以代码处理这种情况并选择一个有效的 Close 系列。
如果数据缺失、为空或不包含可用的 Close 列，机器人会打印错误并跳过该周期。如果数据看起来良好，它将继续并添加指标。
4、添加 EMA、RSI、MACD 和布林带
一旦清洁的价格数据准备好，机器人会添加技术指标。首先，它在 Close 价格上计算 EMA 9 和 EMA 21。这两条线显示短期方向并帮助检测交叉。
接下来，它使用基于指数移动平均的平滑收益和损失公式计算 RSI。这会给出一个介于 0 和 100 之间的值，显示市场目前是上涨还是下跌，以及是否进入更极端的区域。
然后，机器人通过从快速 EMA 中减去慢速 EMA 来计算 MACD，构建信号线，并获取 MACD 直方图作为它们之间的差异。直方图显示动量有多强以及方向如何。
最后，它通过取移动平均线并加上或减去标准差的倍数来计算布林带。机器人存储中间带、上带和下带。这使得策略能够看到价格是否被压在近期范围的顶部或底部，并避免在极端波动中买入。
所有指标添加完成后，脚本删除具有缺失值的行，因此每根剩余的蜡烛都有完整的特征集。
5、交易时间和每日总结
机器人使用迪拜时间作为其主要时钟。它只在周一至周五的 11:00 至 22:00 迪拜时间之间交易。如果当前时间在此窗口之外或如果是周末，机器人只是打印一条消息并不会寻找新的交易。
该脚本还跟踪每日表现。它存储日期、胜利次数和失败次数。每次循环运行时，它都会检查迪拜的日期是否发生变化。如果新的一天开始了，机器人会向 Telegram 发送前一天的每日总结，然后重置新一天的计数器。
6、使用五分钟趋势过滤器
这个机器人的一项主要改进是更高时间框架的趋势过滤器。机器人将一分钟的收盘价重新采样为五分钟的蜡烛，然后对重新采样的系列应用 20 个周期的 EMA。如果最后五分钟的收盘价高于这个 EMA，趋势被认为是上升的。如果低于，则认为是下降的。如果没有足够的五分钟数据来计算稳定的 EMA，该函数返回 nothing 并且机器人跳过该次交易。
信号逻辑尊重这个趋势过滤器。只有在五分钟趋势向上时，机器人才允许发出 CALL 信号；只有在五分钟趋势向下时，机器人才允许发出 PUT 信号。这有助于避免与市场主要方向相反的交易。
7、蜡烛实体过滤器和时间过滤器
在机器人决定信号之前，它还会检查当前蜡烛实体的大小。如果 Open 列可用，它会测量 Open 和 Close 的绝对差，并将其与最近二十个蜡烛的平均实体大小进行比较。如果当前实体相对于最近的平均值太小，机器人会将其视为能量不足或犹豫不决的蜡烛并跳过。
时间过滤器随后检查交易窗口。如果不在我之前定义的迪拜时间范围内，机器人会忽略任何潜在的信号。这确保了所有交易符合特定的日常安排，而不是全天候运行。
8、信号逻辑的工作方式
当时间合适且数据准备就绪时，机器人会查看 DataFrame 中的最后两根蜡烛。它读取当前和前一个 EMA 快速和慢速值，当前 RSI，当前 MACD 直方图，当前布林带上下带和当前价格。
对于可能的 CALL 设置，快速 EMA 必须在最新蜡烛上交叉高于慢速 EMA，而在前一根蜡烛上低于或等于。五分钟趋势必须向上。RSI 必须处于健康的牛市范围内，不要太低，也不要极度超买。MACD 直方图必须为正，这表示向上动量。最后，当前价格必须低于上布林带，这样机器人就不会在已经涨到范围顶部的尖峰中买入。
对于可能的 PUT 设置，条件是镜像的。快速 EMA 必须交叉低于慢速 EMA。五分钟趋势必须向下。RSI 必须位于合理的熊市范围内，不要太高的，也不要深度超卖。MACD 直方图必须为负，这表示向下动量。价格必须高于下布林带，这样机器人就不会在价格已经暴跌到底部后卖出。
如果这些完整条件集中的任何一个都没有满足，机器人会打印一条消息，说明这一根蜡烛没有信号，并且什么也不做。
9、Gemini AI 过滤器
当基于规则的信号出现并通过所有条件时，机器人会构建一个紧凑的快照给 Gemini。它会获取最后五十根一分钟蜡烛，并记录三个列表：最近的收盘价、最近的 RSI 值和最近的 MACD 直方图值。它还包括信号的方向、五分钟趋势、当前价格、当前 RSI、当前 MACD 直方图、当前布林带上下带、当前蜡烛实体大小、最近的平均实体大小和迪拜时间。
这个快照会被转换成一个提示，发送给 Gemini 模型。提示告诉 Gemini，这是一个用于基于规则的机器人的教学用的过滤器。它不允许提供财务建议、谈论真实资金或建议仓位大小。它只能做一个决定：APPROVE 或 SKIP。它必须在第一行输出这个决定，然后在第二行给出一个非常简短的解释，说明环境看起来是否干净且方向明确，或者混乱且杂乱无章。
如果 Gemini 被禁用或调用失败，函数默认返回 APPROVE。这样机器人就像一个正常的基于规则的系统一样运行。如果 Gemini 返回 SKIP，机器人会忽略信号并不开仓。如果 Gemini 返回 APPROVE，机器人会继续正常的入场逻辑并将信号发送到 Telegram。
10、跟踪交易和评估结果
每当信号最终被批准时，机器人会将交易存储在一个列表中。每个条目包含蜡烛时间、入场价格、方向（CALL 或 PUT）以及交易打开的迪拜时间。在 DataFrame 中，机器人定义了到期时间为一分钟后的一个固定数量的蜡烛。在设置中，这个值被设为 5，因此逻辑上的到期时间是在入场蜡烛后的五个一分钟后。
交易的评估发生在检查新信号的同一函数中。然而，由于 ONE_SIGNAL_PER_HOUR 是开启的，该函数只在迪拜当前分钟为零时运行其主要逻辑。这意味着实际上，机器人围绕每个小时的顶部对交易评估进行了分组。代码仍然使用 DataFrame 中的五个蜡烛距离来决定比较哪个价格，但实际的检查是在下一个小时周期中进行的，此时加载了新的数据。
当有足够的数据覆盖到期蜡烛时，机器人会将退出价格与入场价格进行比较。对于 CALL 交易，如果退出价格高于入场价则计为胜利。对于 PUT 交易，如果退出价格低于入场价则计为胜利。然后它会发送一条详细的信息到 Telegram，包含入场价格和时间、退出价格和时间以及结果。根据这个结果，它更新当天的胜利或失败计数器。
import os
import time
from datetime import datetime, timedelta, date
import requests
import numpy as np
import pandas as pd
import yfinance as yf
import pytz
import traceback
# 尝试导入 Gemini 客户端
try:
import google.generativeai as genai
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
if GEMINI_API_KEY:
genai.configure(api_key=GEMINI_API_KEY)
GEMINI_ENABLED = True
print("Gemini AI 过滤器已启用。")
else:
GEMINI_ENABLED = False
print("GEMINI_API_KEY 未设置。Gemini AI 过滤器已禁用。")
except ImportError:
GEMINI_ENABLED = False
print("google-generativeai 未安装。Gemini AI 过滤器已禁用。")
# ========== 配置 ==========
PAIR = "EURUSD=X" # 在 Yahoo Finance 上的 EURUSD
INTERVAL = "1m" # 1 分钟蜡烛
EMA_FAST = 9
EMA_SLOW = 21
RSI_PERIOD = 14
MACD_FAST = 12
MACD_SLOW = 26
MACD_SIGNAL = 9
BB_PERIOD = 20
BB_STD = 2
CHECK_EVERY_SECONDS = 60 # 每分钟检查一次
EXIT_CANDLES = 5 # 到期需要多少个 1m 蜡烛 (5 分钟)
ONE_SIGNAL_PER_HOUR = True # 如果为 True，则只在每小时开始时允许信号
# ======== Telegram 设置 ========
TELEGRAM_BOT_TOKEN = "yours"
TELEGRAM_CHAT_ID = "yourchatid"
# =============================
dubai_tz = pytz.timezone("Asia/Dubai")
# 全局状态
last_candle_time = None
open_trades = [] # 字典列表: {entry_time, entry_price, direction, dubai_time_open}
# 每日统计
stats_date = None # 类型: date
wins_today = 0
losses_today = 0
# ========== Telegram 辅助程序 ==========
def send_telegram_message(text: str):
"""
向您的 Telegram 聊天发送纯文本消息。
"""
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": text
}
try:
r = requests.post(url, json=payload, timeout=10)
print("Telegram 状态:", r.status_code, "-", r.text)
if r.status_code != 200:
print("发送 Telegram 消息出错:", r.text)
except Exception as e:
print("Telegram 发送错误:", e)
# ========== 数据获取 ==========
def fetch_data() -> pd.DataFrame:
"""
从 Yahoo Finance 获取最近的 EURUSD 蜡烛。
标准化列以确保始终有一个可用的 'Close' 列。
"""
try:
df = yf.download(
PAIR,
period="2d",
interval=INTERVAL,
auto_adjust=False,
progress=False
)
except Exception as e:
print("从 Yahoo Finance 获取数据出错:", e)
return pd.DataFrame()
if df is None:
print("Yahoo Finance 返回 None")
return pd.DataFrame()
if isinstance(df, pd.Series):
df = df.to_frame("Close")
if not isinstance(df, pd.DataFrame):
print("从 yfinance 得到意外的数据类型:", type(df))
return pd.DataFrame()
if df.empty:
print("Yahoo Finance 返回空 DataFrame")
return pd.DataFrame()
# 如果列是 MultiIndex，选择一个 Close 列
if isinstance(df.columns, pd.MultiIndex):
if ("Close", PAIR) in df.columns:
df = df.copy()
df["Close"] = df[("Close", PAIR)]
else:
close_cols = [c for c in df.columns if c[0] == "Close"]
if close_cols:
df = df.copy()
df["Close"] = df[close_cols[0]]
else:
print("无法在 MultiIndex 数据中找到 Close 列")
return pd.DataFrame()
else:
if "Close" not in df.columns:
print("数据中没有 Close 列")
return pd.DataFrame()
df = df.dropna()
return df
# ========== 指标 ==========
def compute_rsi(close: pd.Series, period: int) -> pd.Series:
delta = close.diff()
gain = delta.clip(lower=0)
loss = -delta.clip(upper=0)
avg_gain = gain.ewm(alpha=1 / period, adjust=False).mean()
avg_loss = loss.ewm(alpha=1 / period, adjust=False).mean()
rs = avg_gain / (avg_loss + 1e-10)
rsi = 100 - (100 / (1 + rs))
return rsi
def compute_macd(close: pd.Series, fast: int, slow: int, signal: int):
ema_fast = close.ewm(span=fast, adjust=False).mean()
ema_slow = close.ewm(span=slow, adjust=False).mean()
macd = ema_fast - ema_slow
signal_line = macd.ewm(span=signal, adjust=False).mean()
hist = macd - signal_line
return macd, signal_line, hist
def compute_bollinger(close: pd.Series, period: int, std_mult: float):
ma = close.rolling(period).mean()
std = close.rolling(period).std()
upper = ma + std_mult * std
lower = ma - std_mult * std
return ma, upper, lower
def add_indicators(df: pd.DataFrame) -> pd.DataFrame:
"""
在 DataFrame 中添加 EMA 9、EMA 21、RSI、MACD、布林带。
"""
if df is None or df.empty:
return pd.DataFrame()
df = df.copy()
close_raw = df["Close"]
# 如果 Close 是 DataFrame (罕见情况)，取第一列
if isinstance(close_raw, pd.DataFrame):
close_col = close_raw.iloc[:, 0]
else:
close_col = close_raw
close = pd.to_numeric(close_col, errors="coerce")
close = pd.Series(close, index=df.index)
# EMAs
ema_fast = close.ewm(span=EMA_FAST, adjust=False).mean()
ema_slow = close.ewm(span=EMA_SLOW, adjust=False).mean()
df["ema_fast"] = ema_fast
df["ema_slow"] = ema_slow
# RSI
df["rsi"] = compute_rsi(close, RSI_PERIOD)
# MACD
macd, macd_signal, macd_hist = compute_macd(close, MACD_FAST, MACD_SLOW, MACD_SIGNAL)
df["macd"] = macd
df["macd_signal"] = macd_signal
df["macd_hist"] = macd_hist
# 布林带
bb_ma, bb_upper, bb_lower = compute_bollinger(close, BB_PERIOD, BB_STD)
df["bb_ma"] = bb_ma
df["bb_upper"] = bb_upper
df["bb_lower"] = bb_lower
# 删除具有任何 NaN 的行
df = df.dropna()
return df
# ========== 每日统计 ==========
def check_and_reset_daily_stats():
"""
如果迪拜日期改变，发送前一天的摘要并重置计数器。
"""
global stats_date, wins_today, losses_today
now_dubai = datetime.now(dubai_tz)
today = now_dubai.date()
if stats_date is None:
# 第一次初始化
stats_date = today
wins_today = 0
losses_today = 0
return
if today != stats_date:
# 新的一天开始，发送前一天的摘要
summary = (
f"📊 前一天的每日摘要 {stats_date.isoformat()} (迪拜时间)\n"
f"胜利: {wins_today}\n"
f"失败: {losses_today}"
)
print(summary)
send_telegram_message(summary)
# 重置新一天
stats_date = today
wins_today = 0
losses_today = 0
# ========== 时间过滤器 ==========
def is_good_trading_time(now_dubai: datetime) -> bool:
"""
仅在主要外汇时间交易:
周一至周五，迪拜时间 11:00 至 22:00。
"""
weekday = now_dubai.weekday() # 0 = 星期一，6 = 星期日
if weekday >= 5:
return False
hour = now_dubai.hour
return 11 <= hour < 22
# ========== 五分钟趋势 ==========
def get_5m_trend(df: pd.DataFrame) -> str:
"""
使用重新采样的五分钟蜡烛上的 EMA 确定五分钟趋势。
返回 "UP"、"DOWN" 或 None。
"""
if df is None or df.empty:
return None
# 使用 Close 进行五分钟重新采样
close_1m = df["Close"]
if isinstance(close_1m, pd.DataFrame):
close_1m = close_1m.iloc[:, 0]
close_1m = pd.to_numeric(close_1m, errors="coerce")
df_5m = close_1m.resample("5T").last().dropna()
if len(df_5m) < 30:
return None
ema_5m = df_5m.ewm(span=20, adjust=False).mean()
last_close = df_5m.iloc[-1]
last_ema = ema_5m.iloc[-1]
if last_close > last_ema:
return "UP"
elif last_close < last_ema:
return "DOWN"
else:
return None
# ========== 交易评估 ==========
def evaluate_open_trades(df: pd.DataFrame):
"""
检查是否有任何开放交易已完成 EXIT_CANDLES 蜡烛，
然后发送 WIN 或 LOSS 结果并更新每日统计。
"""
global open_trades, wins_today, losses_today
if df is None or df.empty:
return
if len(open_trades) == 0:
return
index_list = list(df.index)
trades_to_remove = []
for trade in open_trades:
entry_time = trade["entry_time"]
direction = trade["direction"]
entry_price = float(trade["entry_price"])
if entry_time not in df.index:
continue
try:
entry_idx = index_list.index(entry_time)
except ValueError:
continue
exit_idx = entry_idx + EXIT_CANDLES # 在 1 分钟图表上到期
if exit_idx >= len(df):
continue
exit_candle = df.iloc[exit_idx]
exit_price = float(exit_candle["Close"])
# 决定赢或输
if direction == "CALL":
result = "WIN" if exit_price > entry_price else "LOSS"
else: # PUT
result = "WIN" if exit_price < entry_price else "LOSS"
dubai_time_open = trade["dubai_time_open"]
dubai_time_close = dubai_time_open + timedelta(minutes=EXIT_CANDLES)
msg = (
f"关于 {direction} EURUSD 交易的结果\n"
f"入场价格: {entry_price:.5f} 在 {dubai_time_open.strftime('%Y-%m-%d %H:%M:%S')} 迪拜时间\n"
f"在 {EXIT_CANDLES} 分钟后退出: {exit_price:.5f} 在 {dubai_time_close.strftime('%Y-%m-%d %H:%M:%S')} 迪拜时间\n"
f"结果: {result}"
)
print(msg)
send_telegram_message(msg)
# 更新统计
if result == "WIN":
wins_today += 1
else:
losses_today += 1
trades_to_remove.append(trade)
for t in trades_to_remove:
if t in open_trades:
open_trades.remove(t)
# ========== Gemini AI 过滤器辅助程序 ==========
def build_ai_snapshot(df: pd.DataFrame,
direction: str,
trend_5m: str,
rsi_now: float,
macd_hist_now: float,
price_now: float,
bb_upper_now: float,
bb_lower_now: float,
body_size: float,
avg_body: float,
now_dubai: datetime) -> dict:
"""
为 AI 过滤器构建最近市场状况的紧凑快照。
我们保持它小巧，以便提示轻量。
"""
recent = df.tail(50).copy()
# 限制为浮点数，保留合理的精度以避免巨大的字符串
closes = [float(x) for x in recent["Close"].round(5).tolist()]
rsis = [float(x) for x in recent["rsi"].round(2).tolist()]
macd_hists = [float(x) for x in recent["macd_hist"].round(6).tolist()]
snapshot = {
"direction": direction,
"trend_5m": trend_5m,
"price_now": float(price_now),
"rsi_now": float(rsi_now),
"macd_hist_now": float(macd_hist_now),
"bb_upper_now": float(bb_upper_now),
"bb_lower_now": float(bb_lower_now),
"body_size": float(body_size),
"avg_body": float(avg_body),
"recent_closes": closes,
"recent_rsi": rsis,
"recent_macd_hist": macd_hists,
"time_dubai": now_dubai.strftime('%Y-%m-%d %H:%M:%S'),
}
return snapshot
def ask_gemini_should_trade(snapshot: dict) -> tuple[str, str]:
"""
询问 Gemini 是否应 APPROVE 或 SKIP 此信号。
返回:
(decision, explanation)
- decision: "APPROVE" 或 "SKIP"
- explanation: 简短的人类可读原因
如果 Gemini 被禁用或出错，我们默认返回 APPROVE，这样机器人像以前一样运行。
这仅用于教育/演示用途，不是财务建议。
"""
if not GEMINI_ENABLED:
return "APPROVE", "Gemini 过滤器已禁用。"
prompt = f"""
你是一个教育交易助手，帮助过滤来自基于规则的 EURUSD 机器人的信号。
你**不允许**提供财务建议、利润保证或仓位规模指导。
你只决定这个信号看起来是否足够干净可以考虑，或者应该被跳过。
数据（最近的 EURUSD 条件）：
方向: {snapshot['direction']} # CALL 表示买入，PUT 表示卖出
五分钟趋势方向（来自 EMA）: {snapshot['trend_5m']}
当前价格: {snapshot['price_now']}
当前 RSI (1m): {snapshot['rsi_now']}
当前 MACD 直方图 (1m): {snapshot['macd_hist_now']}
当前布林带上限: {snapshot['bb_upper_now']}
当前布林带下限: {snapshot['bb_lower_now']}
当前蜡烛实体大小: {snapshot['body_size']} (最近 20 个实体的平均值: {snapshot['avg_body']})
本地时间 (迪拜): {snapshot['time_dubai']}
最近的收盘价 (1m，最近的最后，最多 50 个值):
{snapshot['recent_closes']}
最近的 RSI 值 (1m):
{snapshot['recent_rsi']}
最近的 MACD 直方图值 (1m):
{snapshot['recent_macd_hist']}
你的任务:
1. 决定这个信号看起来是否处于相对干净、方向明确的环境，还是混乱、不稳定或范围内的环境。
2. 保持保守。如果你不确定，优先跳过。
3. 输出正好两行:
- 第一行: 无论是 APPROVE 还是 SKIP (大写)。
- 第二行: 一个简短的句子解释你的决定，从教育的角度。
记住:
- 不要提及真实资金。
- 不要告诉用户进行交易、调整仓位或承诺好的结果。
- 你的答案只是一个教育过滤器，叠加在现有的基于规则的机器人之上。
"""
try:
model = genai.GenerativeModel("gemini-1.5-flash")
response = model.generate_content(prompt)
text = (response.text or "").strip()
if not text:
return "APPROVE", "AI 返回空响应，默认为 APPROVE。"
lines = [line.strip() for line in text.splitlines() if line.strip()]
decision = lines[0].upper() if lines else "APPROVE"
if decision not in ("APPROVE", "SKIP"):
decision = "APPROVE"
explanation = lines[1] if len(lines) > 1 else "AI 没有解释。"
return decision, explanation
except Exception as e:
print("Gemini 在 ask_gemini_should_trade 中出错:", repr(e))
return "APPROVE", "AI 过滤器出错，默认为 APPROVE。"
# ========== 信号逻辑 ==========
def format_dubai_time_short(dt: datetime) -> str:
"""
格式化时间如 2:55pm (没有前导零，小写 am/pm)。
"""
return dt.strftime("%I:%M%p").lstrip("0").lower()
def check_for_signal():
"""
每根蜡烛的逻辑:
- 检查每日重置
- 获取数据
- 添加指标
- 评估现有交易
- 检查新多过滤器信号 (CALL/PUT)
- 可选: 询问 Gemini AI 是否应 APPROVE 或 SKIP
- 每小时模式: 如果 ONE_SIGNAL_PER_HOUR 为 True，则只在每小时开始时触发信号
"""
global last_candle_time, open_trades
# 1) 每日摘要检查
check_and_reset_daily_stats()
now_dubai = datetime.now(dubai_tz)
# 仅在每小时开始时允许信号，如果启用
if ONE_SIGNAL_PER_HOUR:
if now_dubai.minute != 0:
print(f"{now_dubai} - 不是每小时开始，跳过信号检查。")
return
# 2) 获取数据
df = fetch_data()
if df is None or df.empty:
print(f"{now_dubai} - 从 Yahoo Finance 未收到数据")
return
# 3) 指标
df = add_indicators(df)
if df is None or df.empty:
print(f"{now_dubai} - 指标后数据不足")
return
# 4) 评估之前的开放交易
evaluate_open_trades(df)
# 5) 需要至少 3 根蜡烛
if len(df) < 3:
print(f"{now_dubai} - 信号检查蜡烛不足")
return
latest = df.iloc[-1]
prev = df.iloc[-2]
candle_time = latest.name
price_now = float(latest["Close"])
ema_fast_now = float(latest["ema_fast"])
ema_slow_now = float(latest["ema_slow"])
ema_fast_prev = float(prev["ema_fast"])
ema_slow_prev = float(prev["ema_slow"])
rsi_now = float(latest["rsi"])
macd_hist_now = float(latest["macd_hist"])
bb_upper_now = float(latest["bb_upper"])
bb_lower_now = float(latest["bb_lower"])
# 蜡烛实体大小过滤器 (固定为浮点数)
if "Open" in df.columns:
open_now = float(latest["Open"])
body_series = (df["Close"] - df["Open"]).abs()
avg_body = float(body_series.tail(20).mean())
else:
open_now = price_now
avg_body = 0.0
body_size = abs(price_now - open_now)
# 如果 avg_body 是零或 NaN，回退到当前实体大小
if not np.isfinite(avg_body) or avg_body == 0:
avg_body = body_size
# 避免微小的十字蜡烛 (太弱)
if avg_body > 0 and body_size < 0.3 * avg_body:
print(f"{now_dubai} - 蜡烛太小 (实体 {body_size:.6f} vs 平均 {avg_body:.6f}), 跳过。")
return
# 避免在非交易时间交易
if not is_good_trading_time(now_dubai):
print(f"{now_dubai} - 在主要交易时间之外，跳过交易。")
return
# 6) 五分钟趋势过滤器
trend_5m = get_5m_trend(df)
if trend_5m is None:
print(f"{now_dubai} - 五分钟趋势不明确，跳过。")
return
# 避免在同一根蜡烛上重复信号
if last_candle_time is not None and candle_time <= last_candle_time:
print(f"{now_dubai} - 尚未有新蜡烛 (上次蜡烛时间 {last_candle_time})。")
return
signal_text = None
direction = None
reason_parts = []
# 基础 EMA 交叉条件
call_cross = (ema_fast_prev <= ema_slow_prev) and (ema_fast_now > ema_slow_now)
put_cross = (ema_fast_prev >= ema_slow_prev) and (ema_fast_now < ema_slow_now)
# === CALL 条件 ===
if call_cross:
# 趋势过滤器: 仅在上涨趋势中买入
if trend_5m != "UP":
print(f"{now_dubai} - CALL 交叉但五分钟趋势不是 UP，跳过。")
# RSI: 避免超买; 想要牛市但不极端
elif not (40 <= rsi_now <= 70):
print(f"{now_dubai} - CALL 交叉但 RSI={rsi_now:.2f} 在 40-70 之外，跳过。")
# MACD: 需要牛市动量
elif macd_hist_now <= 0:
print(f"{now_dubai} - CALL 交叉但 MACD 直方图 {macd_hist_now:.6f} <= 0，跳过。")
# 布林带: 避免在或高于上轨的尖峰
elif price_now >= bb_upper_now:
print(f"{now_dubai} - CALL 交叉但价格在或高于上布林带，跳过。")
else:
signal_text = "买入 (CALL)"
direction = "CALL"
reason_parts.append("EMA9 交叉高于 EMA21")
reason_parts.append("五分钟趋势 UP")
reason_parts.append(f"RSI={rsi_now:.1f} 在 40-70")
reason_parts.append("MACD 动量牛市")
reason_parts.append("价格低于上布林带")
# === PUT 条件 ===
if direction is None and put_cross:
# 趋势过滤器: 仅在下跌趋势中卖出
if trend_5m != "DOWN":
print(f"{now_dubai} - PUT 交叉但五分钟趋势不是 DOWN，跳过。")
# RSI: 避免超卖; 想要熊市但不极端
elif not (30 <= rsi_now <= 60):
print(f"{now_dubai} - PUT 交叉但 RSI={rsi_now:.2f} 在 30-60 之外，跳过。")
# MACD: 需要熊市动量
elif macd_hist_now >= 0:
print(f"{now_dubai} - PUT 交叉但 MACD 直方图 {macd_hist_now:.6f} >= 0，跳过。")
# 布林带: 避免在或低于下轨的尖峰
elif price_now <= bb_lower_now:
print(f"{now_dubai} - PUT 交叉但价格在或低于下布林带，跳过。")
else:
signal_text = "卖出 (PUT)"
direction = "PUT"
reason_parts.append("EMA9 交叉低于 EMA21")
reason_parts.append("五分钟趋势 DOWN")
reason_parts.append(f"RSI={rsi_now:.1f} 在 30-60")
reason_parts.append("MACD 动量熊市")
reason_parts.append("价格高于下布林带")
# ========== AI 过滤器 + 最终信号 ==========
if signal_text is not None and direction is not None:
# 构建 AI 快照
ai_snapshot = build_ai_snapshot(
df=df,
direction=direction,
trend_5m=trend_5m,
rsi_now=rsi_now,
macd_hist_now=macd_hist_now,
price_now=price_now,
bb_upper_now=bb_upper_now,
bb_lower_now=bb_lower_now,
body_size=body_size,
avg_body=avg_body,
now_dubai=now_dubai
)
ai_decision, ai_reason = ask_gemini_should_trade(ai_snapshot)
if ai_decision != "APPROVE":
print(f"{now_dubai} - AI 过滤器跳过 {direction} 信号。原因: {ai_reason}")
# 如果你想在 Telegram 中看到 AI 跳过，请取消注释:
# send_telegram_message(f"AI 过滤器跳过了一个 {direction} 信号。原因: {ai_reason}")
return
# AI 批准，继续正常
last_candle_time = candle_time
dubai_entry_time = now_dubai
dubai_expiry_time = dubai_entry_time + timedelta(minutes=EXIT_CANDLES)
open_trades.append({
"entry_time": candle_time,
"entry_price": price_now,
"direction": direction,
"dubai_time_open": dubai_entry_time
})
first_line = (
f"{signal_text} EURUSD (每小时信号) 在 "
f"{format_dubai_time_short(dubai_entry_time)} 迪拜时间进行 {EXIT_CANDLES} 分钟"
)
reason_text = "; ".join(reason_parts)
msg = (
f"{first_line}\n\n"
f"入场价格: {price_now:.5f}\n"
f"入场时间: {dubai_entry_time.strftime('%Y-%m-%d %H:%M:%S')} 迪拜时间\n"
f"到期: {dubai_expiry_time.strftime('%Y-%m-%d %H:%M:%S')} 迪拜时间\n"
f"原因: {reason_text}\n\n"
f"AI 过滤器: {ai_decision} - {ai_reason}"
)
print(msg)
send_telegram_message(msg)
else:
print(f"{now_dubai} - 本根蜡烛没有信号。")
# ========== 主循环 ==========
def main_loop():
print("正在启动信号机器人 (带有可选 Gemini 过滤器的高级策略，每小时信号)...")
send_telegram_message(
"✅ 高级 EURUSD 信号机器人已启动 "
"(EMA+RSI+MACD+BB+5m 趋势 + 可选 Gemini 过滤器，每小时信号)。仅用于演示/教育用途。"
)
while True:
try:
check_for_signal()
except Exception as e:
print("主循环出错:", repr(e))
traceback.print_exc()
time.sleep(CHECK_EVERY_SECONDS)
if __name__ == "__main__":
main_loop()
11、结束语
这个机器人并不是用来预测市场的或保证结果的。它只是一个简单的例子，展示了 Python、简单的指标、更高时间框架的趋势检查和轻量级 AI 过滤器如何协同工作。目标是减少噪音，保持一致性，并研究基于规则的逻辑在实时中的行为。如果你决定自己构建类似的东西，请保持简单，花时间，并把它当作一个学习项目。
原文链接：I Built an Advanced EURUSD Signal Bot With EMA, RSI, MACD, Bollinger Bands and a Gemini AI Filter
DefiPlot翻译整理，转载请标明出处
免责声明：本站资源仅用于学习目的，也不应被视为投资建议，读者在采取任何行动之前应自行研究并对自己的决定承担全部责任。
