从零开始!手把手教你搭建一个会”思考”的外汇交易AI机器人（附源码）

作者：老余捞鱼

原创不易，转载请标明出处及原作者。

写在前面的话：今天跟大家分享我最新搭建的欧美外汇自动交易机器人。这套系统整合了EMA均线、RSI强弱指标、MACD趋势线和布林带四大经典武器，还加入了谷歌Gemini AI智能过滤层，能自动识别市场陷阱。从数据抓取到信号生成，从风险控制到自动下单，全程Python代码实现。文章包含完整代码教学，即使是编程小白也能跟着上手。

一、为什么我要做这个交易机器人?

你有没有在做外汇时遇到过这些情况？明明技术分析显示该止损了，但心里总想着”再等等，说不定能回本”；或者看到价格突破了，激动地冲进去，结果发现是假突破，被套在山顶。这就是人性的弱点：贪婪和恐惧。

那能不能让机器来替我们交易？让冰冷的代码执行纪律，让数据说话，让AI帮我们过滤掉那些看起来诱人实际是陷阱的信号?

今天分享给大家的这套欧美外汇自动交易机器人，它整合了四大经典技术指标(EMA均线、RSI强弱指标、MACD趋势线、布林带),还加入了谷歌Gemini AI作为最后一道智能过滤关卡，并可以通过消息网关直接通知交易情况。

💡 核心亮点:

• 四重技术指标过滤，信号准确率大幅提升。
• 5分钟高周期趋势判断，只做顺势单。
• Gemini AI智能过滤，自动识别市场陷阱。
• 全自动执行，从信号生成到下单一气呵成。
• 完整Python代码，可直接运行。

二、技术指标组合拳：四大武器协同作战

在开始讲代码之前，我先给大家科普一下这套系统用到的四大技术指标。如果你是老手，可以快速浏览；如果你是新手，建议认真看，这些都是交易的基本功。

2.1 EMA均线：趋势的”指南针”

EMA(指数移动平均线)是一种对近期价格更敏感的均线。和普通均线(SMA)相比，EMA对最新价格的权重更大，所以能更快地反映市场变化。

我的系统用了两条EMA：

• 快线EMA9：9根K线的指数平均，反应短期趋势。
• 慢线EMA21：21根K线的指数平均，反应中期趋势。

交易逻辑：

• 当快线从下方穿过慢线(金叉)，说明短期动能开始强于中期，可能是上涨信号。
• 当快线从上方穿过慢线(死叉)，说明短期动能开始弱于中期，可能是下跌信号。

2.2 RSI相对强弱指标：市场的”温度计”

RSI(Relative Strength Index)是用来衡量市场超买超卖状态的指标，数值范围是0-100。

判断标准:

• RSI > 70：市场可能超买，价格涨太快，随时可能回调。
• RSI < 30：市场可能超卖，价格跌太狠，随时可能反弹。
• RSI在40-60之间：市场相对健康，适合交易。

我的系统设置:

• 做多信号：RSI在40-70之间(避免追高)
• 做空信号：RSI在30-60之间(避免杀跌)

⚠️ RSI有个致命缺陷叫”钝化”。在超级大牛市或大崩盘中，RSI会长期停留在80以上或20以下。这时候如果你看到RSI80就去做空，可能会被暴涨打脸。所以我加入了趋势过滤,只在趋势方向做单。

2.3 MACD：动量的”加速器”

MACD(平滑异同移动平均线)由三部分组成：

• MACD线：快线EMA12 – 慢线EMA26
• 信号线：MACD线的9日EMA
• 柱状图：MACD线 – 信号线

我的使用方法:

• 做多时：MACD柱状图必须为正(说明上涨动能强)；
• 做空时：MACD柱状图必须为负(说明下跌动能强)。

这样可以避免在动能不足时进场，减少被反向打止损的概率。

2.4 布林带：波动的”边界线”

布林带(Bollinger Bands)由三条线组成：

• 中轨：20日移动平均线；
• 上轨：中轨 + 2倍标准差；
• 下轨：中轨 – 2倍标准差。

根据统计学原理，价格有95%的时间会在上下轨之间波动。

我的使用逻辑:

• 做多时：价格不能在上轨或以上(避免追高被套)；
• 做空时：价格不能在下轨或以下(避免杀跌被反弹)。

2.5 四大指标协同工作流程

单独使用任何一个指标都容易产生假信号，但当四个指标同时满足条件时,信号的可靠性会大大提升。下面是我的做多信号生成条件:

只有当所有条件同时满足时，系统才会生成做多信号。做空信号的逻辑正好相反。

三、5分钟高周期趋势过滤:只做顺势单

很多新手做交易，喜欢在1分钟图上抓小波动，结果经常被来回打脸。这是因为小周期的噪音太多。

我的解决方案是：用5分钟周期判断大趋势，只在趋势方向做单。

3.1 趋势判断逻辑

系统会把1分钟K线重新采样成5分钟K线，然后计算5分钟的20周期EMA。

• 如果5分钟收盘价 > 5分钟EMA20，判断为上升趋势，只允许做多。
• 如果5分钟收盘价 < 5分钟EMA20，判断为下降趋势，只允许做空。

这样做的好处:

• ✅ 避免在震荡市中来回被打；
• ✅ 提高信号的成功率；
• ✅ 符合”顺势而为”的交易哲学；

实战经验: 我测试过很多周期组合，发现1分钟信号+5分钟趋势过滤是最佳搭配。15分钟太慢，3分钟又不够稳定，5分钟刚刚好。

四、Gemini AI智能过滤：机器人的”大脑”

前面讲的四大指标和趋势过滤,都是基于规则的硬性筛选。但市场是复杂的，有时候即使所有指标都满足，信号质量也可能不高。

这时候就需要AI来做最后的判断。

4.1 为什么选择Gemini AI?

我对比过ChatGPT、Claude、DeepSeek、Gemini等多个AI模型，最终选择了谷歌的Gemini Flash 3，原因有三:

1. 推理能力强：能理解复杂的市场数据组合；
2. 响应速度快：延迟低,适合实时交易场景；
3. 免费额度大：对个人开发者友好。

4.2 AI过滤的工作流程

当系统生成一个交易信号后，会做以下操作：

1. 构建市场快照：收集最近50根K线的收盘价、RSI值、MACD柱状图。
2. 生成提示词：把当前信号方向、趋势、指标数值、市场快照打包成一段文字。
3. 询问AI：”这个信号看起来是在干净的趋势环境,还是在震荡混乱的环境?”
4. 获取判断：AI返回”APPROVE”(批准)或”SKIP”(跳过)。
5. 执行决策：只有AI批准的信号才会真正下单。

4.3 AI提示词示例

下面是我给AI的提示词模板(简化版)：

你是一个交易助手,帮助过滤交易信号。当前信号:做多(CALL)5分钟趋势:上升当前价格:1.08500当前RSI:55.2当前MACD柱状图:0.00012布林带上轨:1.08650布林带下轨:1.08350最近50根K线收盘价:[1.0840, 1.0845, 1.0850, ...]最近50根RSI值:[52.1, 53.5, 55.2, ...]最近50根MACD柱状图:[0.00008, 0.00010, 0.00012, ...]请判断这个信号是否处于干净的趋势环境。如果市场看起来清晰、方向明确,请回复"APPROVE"。如果市场看起来混乱、震荡、不确定,请回复"SKIP"。第一行输出:APPROVE 或 SKIP第二行输出:简短理由

4.4 AI过滤的实际效果

我做了一个月的回测对比：

可以看到，虽然AI过滤掉了43%的信号，但剩下的信号质量明显提升，胜率从65%提高到79%，盈亏比也从1.2提升到1.5。

重要提示： AI不是万能的，它也会犯错。我的策略是把AI当作”参谋”，而不是”司令”。最终的交易逻辑还是基于技术指标，AI只是帮我们过滤掉一些低质量信号。

五、完整Python代码实现：手把手教学

理论讲完了，现在进入最激动人心的部分——代码实现！

我会把完整代码分成几个模块来讲解，每个模块都有详细注释。即使你是Python新手，跟着我的步骤也能跑起来。

5.1 环境准备

首先你需要安装以下Python库：

pip install yfinance pandas numpy pytz requests google-generativeai

各个库的作用：

• yfinance：从雅虎财经获取外汇历史数据。
• pandas：数据处理和分析。
• numpy：数值计算。
• pytz：时区处理(我用的是迪拜时间)。
• requests：发送Telegram消息通知。
• google-generativeai：调用Gemini AI。

5.2 配置参数

在代码开头,我们先定义所有的配置参数:

# 交易品种和周期PAIR = "EURUSD=X" # 欧美外汇INTERVAL = "1m" # 1分钟K线# 技术指标参数EMA_FAST = 9 # 快线周期EMA_SLOW = 21 # 慢线周期RSI_PERIOD = 14 # RSI周期MACD_FAST = 12 # MACD快线MACD_SLOW = 26 # MACD慢线MACD_SIGNAL = 9 # MACD信号线BB_PERIOD = 20 # 布林带周期BB_STD = 2 # 布林带标准差倍数# 交易设置CHECK_EVERY_SECONDS = 60 # 每60秒检查一次EXIT_CANDLES = 5 # 持仓5根K线后平仓ONE_SIGNAL_PER_HOUR = True # 每小时只发一个信号# Telegram通知TELEGRAM_BOT_TOKEN = "你的机器人Token"TELEGRAM_CHAT_ID = "你的聊天ID"# Gemini AIGEMINI_API_KEY = "你的Gemini API密钥"

💡 如何获取Telegram Bot Token?

1. 在Telegram搜索@BotFather；
2. 发送/newbot创建机器人；
3. 按提示设置名称,获取Token；
4. 搜索@userinfobot获取你的Chat ID。

5.3 数据获取模块

这个函数负责从雅虎财经获取欧美外汇的历史数据：

import yfinance as yfimport pandas as pddef fetch_data():    """获取EURUSD历史数据"""    try:        # 下载最近2天的1分钟数据        df = yf.download(PAIR, period="2d", interval=INTERVAL, progress=False)                # 处理列名(yfinance返回的是多层索引)        if isinstance(df.columns, pd.MultiIndex):            df = df.copy()            df["Close"] = df[("Close", PAIR)]                df = df.dropna() # 删除空值        return df    except Exception as e:        print(f"数据获取失败: {e}")        return pd.DataFrame()

5.4 技术指标计算模块

这是核心模块，计算EMA、RSI、MACD、布林带：

def compute_rsi(close, period):    """计算RSI指标"""    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 rsidef add_indicators(df):    """添加所有技术指标"""    close = pd.to_numeric(df["Close"], errors="coerce")        # EMA均线    df["ema_fast"] = close.ewm(span=EMA_FAST, adjust=False).mean()    df["ema_slow"] = close.ewm(span=EMA_SLOW, adjust=False).mean()        # RSI    df["rsi"] = compute_rsi(close, RSI_PERIOD)        # MACD    ema_fast = close.ewm(span=MACD_FAST, adjust=False).mean()    ema_slow = close.ewm(span=MACD_SLOW, adjust=False).mean()    df["macd"] = ema_fast - ema_slow    df["macd_signal"] = df["macd"].ewm(span=MACD_SIGNAL, adjust=False).mean()    df["macd_hist"] = df["macd"] - df["macd_signal"]        # 布林带    df["bb_ma"] = close.rolling(BB_PERIOD).mean()    df["bb_std"] = close.rolling(BB_PERIOD).std()    df["bb_upper"] = df["bb_ma"] + BB_STD * df["bb_std"]    df["bb_lower"] = df["bb_ma"] - BB_STD * df["bb_std"]        return df.dropna()

5.5 趋势判断模块

判断5分钟周期的趋势方向:

def get_5m_trend(df):    """判断5分钟趋势"""    close_1m = pd.to_numeric(df["Close"], errors="coerce")        # 重采样为5分钟    df_5m = close_1m.resample("5T").last().dropna()        if len(df_5m) < 30:        return None        # 计算5分钟EMA20    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

5.6 Gemini AI过滤模块

这是最关键的AI判断部分：

import google.generativeai as genaigenai.configure(api_key=GEMINI_API_KEY)def ask_gemini_filter(snapshot):    """询问Gemini AI是否批准信号"""    prompt = f"""你是一个交易助手,帮助过滤信号。当前信号:{snapshot['direction']}5分钟趋势:{snapshot['trend_5m']}当前价格:{snapshot['price_now']}当前RSI:{snapshot['rsi_now']}当前MACD柱状图:{snapshot['macd_hist_now']}最近50根K线收盘价:{snapshot['recent_closes']}最近50根RSI:{snapshot['recent_rsi']}请判断这个信号是否在干净的趋势环境中。第一行输出:APPROVE 或 SKIP第二行输出:简短理由"""    try:        model = genai.GenerativeModel("gemini-1.5-flash")        response = model.generate_content(prompt)        text = response.text.strip()                lines = [line.strip() for line in text.splitlines() if line.strip()]        decision = lines[0].upper() if lines else "APPROVE"        reason = lines[1] if len(lines) > 1 else "无理由"                if decision not in ("APPROVE", "SKIP"):            decision = "APPROVE"                return decision, reason    except Exception as e:        print(f"AI调用失败: {e}")        return "APPROVE", "AI错误,默认批准"

5.7 信号生成主逻辑

这是整个系统的核心，判断是否生成交易信号：

def check_for_signal():    """检查是否有交易信号"""    # 1. 获取数据    df = fetch_data()    if df.empty:        return        # 2. 计算指标    df = add_indicators(df)    if len(df) < 3:        return        # 3. 获取最新数据    latest = df.iloc[-1]    prev = df.iloc[-2]        price = 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 = float(latest["rsi"])    macd_hist = float(latest["macd_hist"])    bb_upper = float(latest["bb_upper"])    bb_lower = float(latest["bb_lower"])        # 4. 判断趋势    trend_5m = get_5m_trend(df)    if trend_5m is None:        return        # 5. 判断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)        signal = None    direction = None        # 6. 做多信号判断    if call_cross:        if trend_5m != "UP":            print("金叉但趋势不向上,跳过")        elif not (40 <= rsi <= 70):            print(f"金叉但RSI={rsi}不在40-70,跳过")        elif macd_hist <= 0:            print("金叉但MACD柱状图为负,跳过")        elif price >= bb_upper:            print("金叉但价格在布林带上轨,跳过")        else:            signal = "做多(CALL)"            direction = "CALL"        # 7. 做空信号判断    if put_cross and signal is None:        if trend_5m != "DOWN":            print("死叉但趋势不向下,跳过")        elif not (30 <= rsi <= 60):            print(f"死叉但RSI={rsi}不在30-60,跳过")        elif macd_hist >= 0:            print("死叉但MACD柱状图为正,跳过")        elif price <= bb_lower:            print("死叉但价格在布林带下轨,跳过")        else:            signal = "做空(PUT)"            direction = "PUT"        # 8. AI过滤    if signal is not None:        snapshot = {            "direction": direction,            "trend_5m": trend_5m,            "price_now": price,            "rsi_now": rsi,            "macd_hist_now": macd_hist,            "recent_closes": df["Close"].tail(50).tolist(),            "recent_rsi": df["rsi"].tail(50).tolist()        }                decision, reason = ask_gemini_filter(snapshot)                if decision == "SKIP":            print(f"AI拒绝信号: {reason}")            return                # 9. 发送通知        message = f"""🚀 {signal} EURUSD进场价格: {price}RSI: {rsi:.1f}MACD柱状图: {macd_hist:.6f}5分钟趋势: {trend_5m}AI判断: {reason}"""        send_telegram(message)        print(message)

5.8 主循环

最后,我们需要一个主循环来不断检查信号：

import timedef main():    print("交易机器人启动...")    send_telegram("✅ 欧美外汇交易机器人已启动")        while True:        try:            check_for_signal()        except Exception as e:            print(f"错误: {e}")                time.sleep(CHECK_EVERY_SECONDS)if __name__ == "__main__":    main()

六、风险管理：让机器人活得更久

再好的策略，如果没有风险管理，也会爆仓。我在系统中加入了多层风险控制：

6.1 K线实体过滤

如果当前K线的实体太小(小于最近20根K线平均实体的30%)，说明市场能量不足,跳过信号。

6.2 交易时段限制

只在周一到周五的11:00-22:00(迪拜时间)交易,避开周末和深夜的低流动性时段。

6.3 每小时信号限制

设置ONE_SIGNAL_PER_HOUR = True，每小时只在整点发信号，避免频繁交易。

6.4 自动止盈止损

每笔交易持仓5根K线(5分钟)后自动平仓，对比进场价和出场价判断盈亏。

6.5 每日统计

系统会记录每天的胜率和盈亏，每天结束时发送统计报告到Telegram。

七、实战效果与优化建议

7.1 我的实盘测试结果

我用这套系统在模拟盘跑了3个月，数据如下:

7.2 优化建议

1. 参数优化：可以用遗传算法或网格搜索找到最优的EMA、RSI、MACD参数组合。
2. 多品种：除了欧美，还可以测试黄金、比特币等品种。
3. 仓位管理：根据信号强度动态调整仓位，强信号加仓，弱信号减仓。
4. 机器学习：用历史数据训练模型，预测信号的成功概率。
5. 情绪指标：加入VIX恐慌指数、资金流向等宏观指标。

八、常见问题解答

Q1: 这套系统能保证盈利吗?

A: 不能。任何交易系统都无法保证100%盈利。这套系统只是提高了胜率，但市场是不可预测的，一定要做好风险管理。

Q2: 我不懂编程，能用吗?

A: 如果你完全不懂编程，建议先学习Python基础。如果你会一点，跟着我的代码注释应该能跑起来。

Q3: Gemini AI会不会很贵?

A: Gemini有免费额度，个人使用完全够了。如果超了，可以升级付费版，价格也不贵。

Q4: 为什么选择欧美外汇?

A: 欧美是全球交易量最大的外汇对，流动性好,点差小，适合量化交易。

Q5: 能直接连接券商自动下单吗?

A: 可以,但需要对接券商的API。不同券商接口不同,需要单独开发。我这套系统目前只做信号提示，手动下单。

九、总结与展望

本文系统介绍了如何使用Python语言搭建一套融合经典技术指标与人工智能过滤的欧美外汇自动交易系统。系统采用模块化设计，涵盖数据获取、指标计算、信号生成、风险管理、交易执行和结果通知全流程,为量化交易爱好者提供了完整的实战参考方案。

• 四重技术过滤协同工作：EMA捕捉趋势、RSI判断超买超卖、MACD确认动量、布林带识别波动。
• 5分钟高周期趋势过滤机制,确保只在大趋势方向交易,有效降低假信号。
• 谷歌Gemini AI智能过滤层,通过分析市场快照数据,自动识别并跳过低质量信号。
• 完整的风险管理体系,包含K线实体过滤、交易时段限制、每小时信号控制和自动止盈止损。
• 全流程Python代码实现,从yfinance数据获取到Telegram消息推送,提供可直接运行的完整方案。

十、完整版源代码下载

我为大家准备了完整的代码和说明，需要的朋友继续请往下看。首先是流程：

然后是代码：

上面是部分截图，需要的留言获得。

 #Python量化交易 #外汇交易机器人 #技术指标策略 #人工智能交易 #自动化交易系统 #EMA均线策略 #RSI指标应用 #MACD趋势分析 #布林带交易 #GeminiAI #量化投资入门 #程序化交易 #金融科技 #算法交易 #交易信号系统

感谢阅读！愿本文为您带来新启发与实用知识。若觉有益，请点赞分享，您的支持是我创作的动力，欢迎留言必复。

风险提示：本文仅供参考，不构成投资建议。量化策略开发应以学习和技术交流为目的。投资有风险，入市需谨慎。