如何通过Pine Script将FXMacroData与TradingView集成

交易视图是大多数零售外汇交易者的选择图表平台,Pine Script是将自定义指标和策略信号带入这些图表中的语言.挑战是PineScript完全运行在 TradingView的沙盒内,它无法在运行时向外部API进行出门HTTP调用.本指南通过两部分的工作流弥合这一空白:一个小Python脚本从FXMacroData获取宏观公告数据,生成可粘贴的Pine script代码,并将结果指标直接将这些数据点覆盖在您的 TradingVview图表上,并使用事件标记,信号行和实时宏观数据表.

步骤1 了解FXMacroData公告终点

在FXMacroData中,每个指标都遵循相同的REST形状. 欧元的政策利率请求看起来像这样:

curl "https://fxmacrodata.com/api/v1/announcements/eur/policy_rate?api_key=YOUR_API_KEY&start=2023-01-01"

JSON响应是一个带有 data 阵列:

{
  "currency": "eur",
  "indicator": "policy_rate",
  "data": [
    {
      "date": "2025-03-06",
      "val": 2.65,
      "announcement_datetime": "2025-03-06T13:15:00Z"
    },
    {
      "date": "2025-01-30",
      "val": 2.90,
      "announcement_datetime": "2025-01-30T13:15:00Z"
    },
    {
      "date": "2024-12-12",
      "val": 3.15,
      "announcement_datetime": "2024-12-12T13:15:00Z"
    }
  ]
}

每个记录都带有一个 date (发布的当地日期:YYY-MM-DD),一个数字 val,以及第二级 UTC announcement_datetime对于可用指标的完整列表,请参阅"全球货币指标" 欧元政策利率文件 查看您的目标货币的指标目录.

步骤2 使用Python获取宏数据

创建一个名为 generate_pine.py. 这个脚本获取所选货币和指标的公告数据,然后将结果串行化成Pine Script数组,然后粘贴到指标文件中.

"""
generate_pine.py — fetch FXMacroData announcements and output Pine Script arrays.

Usage:
    python generate_pine.py --currency eur --indicator policy_rate --start 2023-01-01
"""
import argparse
import json
import os
from datetime import datetime, timezone

import requests

BASE_URL = "https://fxmacrodata.com/api/v1"
API_KEY = os.environ.get("FXMD_API_KEY", "")


def fetch_announcements(currency: str, indicator: str, start: str) -> list[dict]:
    url = f"{BASE_URL}/announcements/{currency}/{indicator}"
    params: dict = {"start": start}
    if API_KEY:
        params["api_key"] = API_KEY
    resp = requests.get(url, params=params, timeout=15)
    resp.raise_for_status()
    return resp.json().get("data", [])


def ts_to_pine_timestamp(iso_str: str) -> int:
    """Convert an ISO-8601 UTC string to a Unix timestamp (milliseconds for Pine)."""
    dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
    return int(dt.timestamp() * 1000)


def build_pine_arrays(records: list[dict]) -> str:
    """Render the data as Pine Script array literals."""
    ts_list = []
    val_list = []
    for r in records:
        ann = r.get("announcement_datetime") or f"{r['date']}T00:00:00Z"
        ts_list.append(str(ts_to_pine_timestamp(ann)))
        val_list.append(str(r["val"]))

    ts_str = ", ".join(ts_list)
    val_str = ", ".join(val_list)
    n = len(records)

    return f"""
// ── Auto-generated by generate_pine.py — do not edit manually ──
// Currency: {records[0].get('currency', '?').upper() if records else '?'}
// Indicator: {records[0].get('indicator', '?') if records else '?'}
// Records: {n}

var int[] _ann_ts  = array.from({ts_str})
var float[] _ann_val = array.from({val_str})
"""


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--currency", required=True)
    parser.add_argument("--indicator", required=True)
    parser.add_argument("--start", default="2023-01-01")
    args = parser.parse_args()

    print(f"Fetching {args.indicator} for {args.currency.upper()} from {args.start} …")
    records = fetch_announcements(args.currency, args.indicator, args.start)
    print(f"  {len(records)} records returned.")
    pine = build_pine_arrays(records)
    print(pine)

    output_file = f"pine_data_{args.currency}_{args.indicator}.txt"
    with open(output_file, "w") as f:
        f.write(pine)
    print(f"Arrays written to {output_file}")


if __name__ == "__main__":
    main()

存储您的API密钥并运行脚本:

export FXMD_API_KEY="YOUR_API_KEY"
python generate_pine.py --currency eur --indicator policy_rate --start 2023-01-01

输出文件 pine_data_eur_policy_rate.txt 您可以运行相同的脚本为CPI (--indicator inflation),PMI (--indicator pmi),或其他任何指标. 汇率数据库现在我们要做什么?

步骤3 建立松字符标志器

打开交易视图,导航到松字符编辑器 (图表底部的"松字编辑"选项卡),并粘贴下面的指示器. 取代 _ann_ts 现在我 _ann_val 列在顶部,由步骤2输出.

// @version=5
// FXMacroData Overlay — policy rate event markers + value line
// Replace the arrays below with output from generate_pine.py

indicator("FXMacroData Macro Overlay", overlay=true, max_labels_count=50, max_lines_count=50)

// ── User inputs ──
i_show_labels   = input.bool(true,  "Show event labels")
i_show_line     = input.bool(true,  "Show macro value line")
i_show_table    = input.bool(true,  "Show data table")
i_label_color   = input.color(color.new(color.orange, 0), "Label colour")
i_line_color    = input.color(color.new(color.orange, 30), "Line colour")

// ── Paste generated arrays here ──
// ---- BEGIN GENERATED SECTION ----
var int[]   _ann_ts  = array.from(1672531200000, 1675209600000, 1677628800000)
var float[] _ann_val = array.from(2.50, 2.65, 3.00)
// ---- END GENERATED SECTION ----

// ── Helper: find the most recent announcement value at or before this bar ──
f_current_val() =>
    float result = na
    int bar_t = time
    for i = 0 to array.size(_ann_ts) - 1
        if array.get(_ann_ts, i) <= bar_t
            result := array.get(_ann_val, i)
            break
    result

// ── Macro overlay line ──
float macro_val = f_current_val()
plot(i_show_line ? macro_val : na, title="Macro value", color=i_line_color,
     linewidth=2, style=plot.style_stepline)

// ── Event marker labels ──
if i_show_labels
    for i = 0 to array.size(_ann_ts) - 1
        int   ts  = array.get(_ann_ts, i)
        float val = array.get(_ann_val, i)
        if ts >= chart.left_visible_bar_time and ts <= chart.right_visible_bar_time
            label.new(
                x       = ts,
                y       = high * 1.002,
                text    = str.tostring(val, "#.##"),
                style   = label.style_diamond,
                color   = i_label_color,
                textcolor = color.white,
                size    = size.small,
                xloc    = xloc.bar_time
            )

// ── Summary table (top-right corner) ──
var table t = table.new(position.top_right, 2, 6,
    bgcolor       = color.new(color.navy, 85),
    border_width  = 1,
    border_color  = color.new(color.gray, 60))

if i_show_table and barstate.islast
    table.cell(t, 0, 0, "Date",  text_color=color.silver, text_size=size.small)
    table.cell(t, 1, 0, "Value", text_color=color.silver, text_size=size.small)
    int rows = math.min(5, array.size(_ann_ts))
    for i = 0 to rows - 1
        int   ts  = array.get(_ann_ts, i)
        float val = array.get(_ann_val, i)
        string date_str = str.format("{0,date,yyyy-MM-dd}", ts)
        table.cell(t, 0, i + 1, date_str, text_color=color.white, text_size=size.tiny)
        table.cell(t, 1, i + 1, str.tostring(val, "#.##"), text_color=color.orange, text_size=size.tiny)

按下 添加到图表. 该指标将覆盖与宏观价值进展相匹配的步骤线,并在每个公告日期下放标记的钻石标记. 滚回历史记录,查看每个政策利率打印与您选择的货币对的价格动作如何一致.

第4步 多个指标中的层

单个指标讲述了部分故事.真正的边缘来自将政策利率,CPI和PMI分层. 对于每个指标重复步骤2,并添加一个额外的部分到松字符,或者创建三个单独的指标并将它们堆叠在同一图表窗口上.

为了一次地获取三个指示符,扩展Python脚本:

INDICATORS = ["policy_rate", "inflation", "pmi"]
CURRENCY   = "eur"
START      = "2023-01-01"

for ind in INDICATORS:
    records = fetch_announcements(CURRENCY, ind, START)
    pine    = build_pine_arrays(records)
    with open(f"pine_data_{CURRENCY}_{ind}.txt", "w") as f:
        f.write(pine)
    print(f"Written pine_data_{CURRENCY}_{ind}.txt  ({len(records)} records)")

每个输出文件都为您提供了一个单独的Pine Script指标的下拉式部分.将每个指标添加到您的图表中,并为每个指数分配一个不同的颜色,例如,政策利率是色,CPI是蓝色,PMI是绿色,这样您就可以一眼看到哪个宏观制度在适用.您可以通过 欧元PMI文件 更多的指标目录.

步骤 5 通过发布日历检查自动更新

在新公告打印时,Pine Script数组中的宏数据会变得过时. 在每次交易会前,通过调用FXMacroData的发布日历终点来自动更新,以检查您上一次再生后是否打印了任何东西. 如果是这样,请重新运行生成器并更新PineScript.

"""
check_and_refresh.py — Re-generate Pine arrays if new announcements have printed.
Run this before each session, e.g. via cron or GitHub Actions.
"""
import subprocess
import requests
import os
from datetime import date, timedelta

BASE_URL = "https://fxmacrodata.com/api/v1"
API_KEY  = os.environ.get("FXMD_API_KEY", "")
PAIRS    = [
    ("eur", "policy_rate"),
    ("eur", "inflation"),
    ("eur", "pmi"),
]

def latest_release_date(currency: str, indicator: str) -> str | None:
    """Return the most recent announcement date for this indicator."""
    params: dict = {"start": str(date.today() - timedelta(days=7))}
    if API_KEY:
        params["api_key"] = API_KEY
    r = requests.get(
        f"{BASE_URL}/announcements/{currency}/{indicator}",
        params=params, timeout=10
    )
    r.raise_for_status()
    data = r.json().get("data", [])
    return data[0]["date"] if data else None

def refresh_all() -> None:
    for currency, indicator in PAIRS:
        latest = latest_release_date(currency, indicator)
        if latest and latest >= str(date.today() - timedelta(days=1)):
            print(f"New print detected: {currency.upper()} {indicator} on {latest}. Regenerating …")
            subprocess.run(
                ["python", "generate_pine.py",
                 "--currency", currency,
                 "--indicator", indicator,
                 "--start", "2023-01-01"],
                check=True,
            )
        else:
            print(f"{currency.upper()} {indicator}: no new prints since yesterday.")

if __name__ == "__main__":
    refresh_all()

计划这个脚本在伦敦开放前的星期一星期五在08:00 UTC运行,使用cron工作,GitHub Actions工作流,或任何任务调度器. 当更新运行时,复制更新的 pine_data_*.txt 您可以在"交易视图"的Pine编辑器中 保存您的图表标记将立即反映最新的公告数据.

步骤6 将 TradingView 警报返回 FXMacroData 语境

工作流程目前推出宏观数据. 进入 步骤6关闭了循环的另一方向:当你的Pine脚本发出一个交易信号 (例如,CPI打印后突破阻力),TradingView的网络链接警报系统将该事件转发到一个小的Flask终点,即刻查询FXMacroData的新环境最新CPI趋势,即将发布的版本和当前COT定位并记录或通知您完整的宏观图像.

创造 webhook_server.py没有

"""
webhook_server.py — Receive TradingView alert webhooks and enrich with FXMacroData context.
Run with: python webhook_server.py
Expose to the internet via ngrok or deploy to a cloud function / VPS.
"""
import os
import json
from datetime import date, timedelta

import requests
from flask import Flask, request, jsonify

app = Flask(__name__)
BASE_URL  = "https://fxmacrodata.com/api/v1"
API_KEY   = os.environ.get("FXMD_API_KEY", "")

def get_macro_context(currency: str) -> dict:
    """Fetch latest policy rate + CPI for the given currency."""
    params: dict = {"start": str(date.today() - timedelta(days=90))}
    if API_KEY:
        params["api_key"] = API_KEY

    ctx: dict = {}
    for indicator in ("policy_rate", "inflation"):
        try:
            r = requests.get(
                f"{BASE_URL}/announcements/{currency}/{indicator}",
                params=params, timeout=10
            )
            r.raise_for_status()
            data = r.json().get("data", [])
            ctx[indicator] = data[0] if data else {}
        except Exception as exc:
            ctx[indicator] = {"error": str(exc)}
    return ctx

@app.route("/tradingview/alert", methods=["POST"])
def receive_alert():
    payload = request.get_json(force=True, silent=True) or {}
    currency = payload.get("currency", "usd").lower()
    signal   = payload.get("signal", "unknown")

    ctx = get_macro_context(currency)
    result = {
        "received_signal": signal,
        "currency": currency.upper(),
        "macro_context": ctx,
    }
    print(json.dumps(result, indent=2))
    return jsonify(result), 200

if __name__ == "__main__":
    app.run(port=5050, debug=False)

安装Flask并运行服务器:

pip install flask
python webhook_server.py

通过网络将本地端口暴露在 鱼没有人知道.

ngrok http 5050

在交易视图中,打开 警报 → 创建警报选择你的Pine脚本条件,启用 网络连接网址,然后粘贴一个结束于 的 ngrok HTTPS URL /tradingview/alert. 将警报消息设置为JSON有效载荷:

{
  "currency": "eur",
  "signal": "breakout_long",
  "ticker": "{{ticker}}",
  "price": {{close}}
}

每次Pine Script条件触发时,TradingView都会将JSON发送到您的终端. Webhook处理器从FXMacroData获取最新的EUR政策利率和CPI,并与信号一起记录完整的宏观语境.您可以扩展此功能,使用丰富的数据发送 Telegram或Slack通知.

步骤7 扩展使用COT定位和FX现货环境

宏观公告数据是图片的一个层. 为了获得更完整的信号,请从FXMacroData添加CFTC交易者承诺定位和外汇现货趋势背景.

欧元期货的PETCH COT定位:

curl "https://fxmacrodata.com/api/v1/cot/eur?api_key=YOUR_API_KEY&start=2024-01-01"

{
  "currency": "eur",
  "data": [
    {
      "date": "2025-03-18",
      "net_noncommercial": 48320,
      "long_noncommercial": 182500,
      "short_noncommercial": 134180
    }
  ]
}

添加一个COT网络定位系列到你的PineScript生成器使用相同的模式. net_noncommercial 指投机交易者是欧元期货的净长期交易者 情绪背景,在结合派的欧洲央行政策信号时增加了信念. 机器人机器人的仪表板 在将其嵌入指标之前.

扩展Python生成器,包括一个COT部分:

def fetch_cot(currency: str, start: str) -> list[dict]:
    url = f"{BASE_URL}/cot/{currency}"
    params: dict = {"start": start}
    if API_KEY:
        params["api_key"] = API_KEY
    resp = requests.get(url, params=params, timeout=15)
    resp.raise_for_status()
    return resp.json().get("data", [])


def build_cot_pine_arrays(records: list[dict]) -> str:
    ts_list  = [str(int(datetime.fromisoformat(r["date"] + "T00:00:00+00:00").timestamp() * 1000)) for r in records]
    net_list = [str(r.get("net_noncommercial", 0)) for r in records]
    ts_str   = ", ".join(ts_list)
    net_str  = ", ".join(net_list)
    return f"""
var int[]   _cot_ts  = array.from({ts_str})
var float[] _cot_net = array.from({net_str})
"""

在松字体,情节 _cot_net 在一个单独的窗口 (设置 overlay=false 您可以看到投机定位趋势,以及价格和宏观公告.