为什么一个早晨准备的代理是你首先可以构建的最高回报率的机器人
如果你是自由选择地交易外汇或是开半自动账本,你一天中最昂贵的45分钟是伦敦开业前的时间. 发布日程通过对哪些指纹进行交叉检查, 美元/日元没有人知道. 欧元/美元没有人知道. 汇率汇率试图记住哪个央行发言人是今天重要.它是重复的,容易出错的,
这正是人工智能代理比人类更好地完成的任务:有限范围,结构化输入,决定性输出. 实时外汇事件代理 运行在交易日之前, 从FXMacroData获取现场事件数据, 根据市场影响排名当天的发布,
让你在任何机器上都能安排一个脚本, 笔记本电脑,树,云虚拟机,
每天在UTC06:30运行的Python代理,查询G10地区未来24小时的FXMacroData发布日历,按影响排名事件,通过LLM总结它们,并将结构化简报推向Telegram或Slack.
预先要求
- 它们是如何实现的?
- 来自FXMacroData的API键 管理API现在我们要做什么?
- 您控制的LLM终点.下列任何一个工作:
- 人类克劳德 (推推理质量).
- 开放AI GPT-4类
- 通过奥拉马进行零成本运行.
- 一个电讯机器人令牌 (通过
@BotFather作为交付道. - 时间安排者.
cron在Linux/macOS上,Windows上的任务调度器,或任何云cron都能做到.
安装依赖:
pip install requests python-dotenv
创建一个 .env 文件里包含你的秘密:
FXMD_API_KEY=your_fxmacrodata_key
LLM_PROVIDER=anthropic # or "openai" or "ollama"
ANTHROPIC_API_KEY=sk-ant-...
TELEGRAM_BOT_TOKEN=...
TELEGRAM_CHAT_ID=...
步骤1:从FXMacroData发布日历中抽取下一个24小时
经纪人的第一项工作是知道实际计划. 其他数据 发布日历终点返回每种货币的计划发布,包括指标名称,计划 UTC 时间,之前的值和重要度排名,如果可用.
查询您交易的每一种主要货币:
curl "https://fxmacrodata.com/api/v1/calendar/usd?api_key=YOUR_API_KEY"
curl "https://fxmacrodata.com/api/v1/calendar/eur?api_key=YOUR_API_KEY"
curl "https://fxmacrodata.com/api/v1/calendar/gbp?api_key=YOUR_API_KEY"
curl "https://fxmacrodata.com/api/v1/calendar/jpy?api_key=YOUR_API_KEY"
包裹这个在Python中,这样代理可以在可配置的货币列表中代:
import os
import requests
from datetime import datetime, timedelta, timezone
from dotenv import load_dotenv
load_dotenv()
API = "https://fxmacrodata.com/api/v1"
KEY = os.environ["FXMD_API_KEY"]
CURRENCIES = ["usd", "eur", "gbp", "jpy", "aud", "cad", "chf", "nzd"]
def fxmd_get(path, **params):
r = requests.get(
f"{API}{path}",
params={"api_key": KEY, **params},
timeout=25,
)
r.raise_for_status()
return r.json()
def upcoming_releases(hours_ahead: int = 24):
now = datetime.now(timezone.utc)
cutoff = now + timedelta(hours=hours_ahead)
releases = []
for ccy in CURRENCIES:
try:
data = fxmd_get(f"/calendar/{ccy}").get("data", [])
except requests.HTTPError:
continue
for ev in data:
try:
ts = datetime.fromisoformat(
ev["announcement_datetime"].replace("Z", "+00:00")
)
except (KeyError, ValueError):
continue
if now <= ts <= cutoff:
releases.append({
"currency": ccy.upper(),
"indicator": ev.get("indicator") or ev.get("name"),
"scheduled_utc": ts.isoformat(),
"prior": ev.get("prior_value"),
"consensus": ev.get("consensus") or ev.get("market_consensus"),
"importance": ev.get("importance") or ev.get("impact"),
})
releases.sort(key=lambda r: r["scheduled_utc"])
return releases
步骤2:按市场可能影响排名发布
没有任何印刷品能动摇市场. 农业以外的工资没有人知道. 核心PCE没有人知道. 美联储政策利率没有人知道. 欧元区的CPI没有人知道. 英国CPI没有 欧洲央行政策利率 对于小规模经济体来说,零售交易的闪电很少重要.
编码一个明确的重量表,这样代理人就不必猜.这是整个系统中最大的杆部分,你的重量,你的边缘.
TIER_1 = {
("USD", "non_farm_payrolls"), ("USD", "core_pce"), ("USD", "policy_rate"),
("USD", "inflation"), ("USD", "fomc_minutes"),
("EUR", "inflation"), ("EUR", "policy_rate"),
("GBP", "inflation"), ("GBP", "policy_rate"),
("JPY", "policy_rate"), ("JPY", "inflation"),
("AUD", "policy_rate"), ("CAD", "policy_rate"),
("CHF", "policy_rate"), ("NZD", "policy_rate"),
}
TIER_2 = {
("USD", "retail_sales"), ("USD", "ism_manufacturing"),
("EUR", "gdp"), ("EUR", "unemployment"),
("GBP", "gdp"), ("GBP", "retail_sales"),
("AUD", "inflation"), ("CAD", "inflation"),
}
def impact_score(event: dict) -> int:
key = (event["currency"], (event["indicator"] or "").lower())
if key in TIER_1:
return 3
if key in TIER_2:
return 2
return 1
def rank(releases):
return sorted(
releases,
key=lambda r: (-impact_score(r), r["scheduled_utc"]),
)
现在你可以将日期分解成一个结构化的列表,
步骤3:添加一夜之间语境,以便代理知道已经移动了什么
一个早晨的简报是不完整的,没有一夜间外汇走势的快照. 拉出最近的现货价格对你关心的对,所以LLM可以编织价格行动到它的叙述.
PAIRS = [("USD", "JPY"), ("EUR", "USD"), ("GBP", "USD"), ("AUD", "USD")]
def overnight_moves():
moves = []
for base, quote in PAIRS:
try:
data = fxmd_get(
"/forex",
base=base,
quote=quote,
).get("data", [])
except requests.HTTPError:
continue
if len(data) < 2:
continue
last = data[-1]["value"]
prev = data[-25]["value"] if len(data) >= 25 else data[0]["value"]
change_pct = (last - prev) / prev * 100
moves.append({
"pair": f"{base}/{quote}",
"last": round(last, 5),
"change_pct_24h": round(change_pct, 2),
})
return moves
如果您还想要定位背景, 添加一个 其他 拉到这里. 保持它是可选的. 简报应该仍然没有它.
步骤4:通过LLM生成简报
代理现在有三个结构化输入:排名事件,一夜间外汇移动,以及当前的UTC时间.将它们通过严格的输出合同传输到模型中,以便每天早上进行分析和一致的简报.
import json
from anthropic import Anthropic
claude = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
SYSTEM_PROMPT = """You are an FX morning-prep analyst.
Given today's ranked releases and overnight moves, produce a briefing that:
- leads with the single most important event of the day,
- groups events by impact tier,
- flags 1-2 specific pairs to watch and why,
- ends with one disciplined risk caveat.
Do not give buy/sell instructions. Stay factual. Max 220 words."""
def generate_briefing(events, moves):
payload = json.dumps({
"utc_now": datetime.now(timezone.utc).isoformat(),
"ranked_events": events,
"overnight_moves": moves,
})
resp = claude.messages.create(
model="claude-3-5-sonnet-latest",
max_tokens=600,
system=SYSTEM_PROMPT,
messages=[{"role": "user", "content": payload}],
)
return resp.content[0].text
这一提示是可靠的:
- 结构化输入. 模型接收JSON,而不是自然语言, 因此它从来不需要从文本中提取日期或数字.
- 硬视野. 系统提示禁止交易建议, 这使得产出即使在波动日也保持一致和有用.
- 长度上限 长时间的简报训练你,这打败了目的.
步骤5:将其发送到 Telegram (或 Slack)
报道必须在早上第一时间到达你已经看过的地方.
def send_telegram(text: str):
token = os.environ["TELEGRAM_BOT_TOKEN"]
chat_id = os.environ["TELEGRAM_CHAT_ID"]
requests.post(
f"https://api.telegram.org/bot{token}/sendMessage",
json={
"chat_id": chat_id,
"text": text,
"parse_mode": "Markdown",
"disable_web_page_preview": True,
},
timeout=15,
)
def run():
events = rank(upcoming_releases(hours_ahead=24))
moves = overnight_moves()
briefing = generate_briefing(events, moves)
header = f"*FX Morning Briefing — {datetime.now(timezone.utc):%a %d %b %Y}*\n\n"
send_telegram(header + briefing)
if __name__ == "__main__":
run()
在 Linux/macOS 上,一个 crontab 输入每周在 06:30 UTC 运行代理:
30 6 * * 1-5 /usr/bin/python3 /opt/fx-agent/morning_brief.py >> /var/log/fx-agent.log 2>&1
简报的实际情况
代理在最近一个CPI周中产生的输出示例:
FX Morning Briefing — Thu 22 May 2026
Today's anchor: US Core PCE at 12:30 UTC. Consensus 0.2% MoM,
prior 0.0%. A second sub-0.1 print would cement the disinflation
narrative; a 0.3+ surprise resets Fed pricing.
Tier 1:
- 12:30 UTC USD Core PCE prior +0.0% cons +0.2%
- 13:30 UTC USD Initial Jobless Claims prior 228k cons 225k
Tier 2:
- 06:00 UTC GBP Retail Sales MoM prior -0.1% cons +0.4%
- 09:00 UTC EUR ECB Minutes (qualitative)
Pairs to watch:
- USD/JPY hovering 158.40 after a quiet Asia. PCE miss → 156s
back in play.
- GBP/USD coiled below 1.2700. Stronger UK retail + soft PCE is
the cleanest setup of the day.
Risk caveat: thin EU liquidity ahead of US data; expect outsized
moves on any surprise print.
这是一个完整的市场预测, 每周每天在90秒内阅读,
在信任它之前,硬化检查清单
- 陈旧的数据保护. 如果任何一级事件错过了预定的时间, 拒绝发送简报.
- 尝试重新使用背面 在 HTTP 短暂故障中. 三次尝试, 2s/4s/8s,然后大声失败.
- 输出验证器 拒绝任何包含"买","卖",或"长"/"短"的法学士答案.
- 警报心跳. 如果代理人未能在 UTC 07:00 之前发送简报,请单独发送自己的ping,以免您默默地失去工作流程.
- 成本上限. 已经设置
max_tokens=600报道每天应该是百分之一.
接下来要去哪里
只有当早晨循环可靠时,
- 内部惊喜警报. 调查 宣布终点 每15分钟,在第1级释放后;当实际离一致值超过值时,请提醒.
- 对于每对人来说, 产生一个更紧. 美元/日元- 只有或 欧元/美元- 只有活跃日版本.
- 位置覆盖. 拉下来 机械设备的数据 并且将其输入到同一个提示器,
- 语音模式 通过TTS模型传递简报, 在你咖啡的时候大声读出来.
获胜不是简报本身,而是你走进交易日的有纪律,可重复的环境. 一旦这种习惯自动化,你建立的其他任何外汇代理基础设施都会放在上面.