파이프 스크립트를 통해 트레이딩뷰와 FXMacroData를 통합하는 방법

트레이딩뷰는 대부분의 소매 FX 트레이더들이 선호하는 차트 플랫폼이며, 파이인 스크립트는 그 차트에 사용자 지정 지표와 전략 신호를 생생하게 만드는 언어입니다. 문제는 파이인스크립트가 완전히 트레이닝뷰의 샌드박스 내부에서 실행된다는 것입니다. 실행 시 외부 API에 출력 HTTP 호출을 할 수 없습니다. 이 가이드는 두 부분 워크플로우로 그 격차를 다룬다. 작은 파이썬 스크립트가 FXMacroData에서 매크로 발표 데이터를 가져오고 붙여넣을 준비가 된 파이인 스кри프트 코드를 생성하고, 생성 된 지표는 해당 데이터 포인트를 이벤트 마커, 신호 라인 및 라이브 매크ሮ 데이터 테이블로 직접 트레이דינג뷰 차트에 덮습니다.

단계 1 FXMacroData 발표 최종 지점을 이해

FXMacroData의 모든 지표는 동일한 REST 형태를 따르고 있습니다. EUR에 대한 정책율 요청은 다음과 같습니다.

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 (출매 지역 날짜 YYYY-MM-DD), 숫자 val, 그리고 두 번째 수준의 UTC announcement_datetime모든 통화와 지표에 걸쳐 일관된 형태가 코드 생성기를 쉽게 만들 수 있습니다. 사용 가능한 지표의 전체 목록을 위해 EUR 정책금리 그리고 목표 통화에 대한 지표 카탈로그를 탐색합니다.

단계 2 파이썬으로 매크로 데이터를 가져오기

이름의 파일을 만들자 generate_pine.py이 스크립트는 선택한 화폐와 지표에 대한 발표 데이터를 가져오고, 그 결과 결과를 파이인 스크립트 배열로 연속화합니다.

"""
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), 또는 다른 지표에서 FXMacroData 카탈로그-

단계 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 단계를 반복하고 파이인 스크립트에 추가 섹션을 추가하거나 세 개의 별도의 지표를 생성하고 동일한 차트 패널에 쌓아 놓습니다.

For a three-indicator fetch in one go, extend the Python script:

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)")

각 출력 파일은 별도의 파이인 스크립트 지표에 대한 드롭 인 섹션을 제공합니다. 각 지표로 각각 추가하고 지표별로 예를 들어 정책율, CPI를 위해 파란색, PMI를 위해 녹색을 구분하여 표시하십시오. EUR PMI 문서 그리고 더 넓은 지표 카탈로그.

단계 5 릴리스 캘린더 확인으로 업데이트를 자동화

파이인 스크립트 배열의 매크로 데이터는 새로운 발표가 인쇄되는 순간 구식화됩니다. 매 거래 세션 전에 FXMacroData의 릴리스 캘린더 엔드포인트에 전화를 걸어 마지막 재생 이후 인쇄된 것이 있는지 확인하여 업데이트를 자동화하십시오. 그렇게되면 생성기를 다시 실행하고 파이인스크립트를 업데이트하십시오.

"""
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에 실행하도록 스케줄링하십시오. 런던 오픈 전에, 크론 작업, GitHub Actions 워크플로우 또는 다른 작업 스케줄러를 사용하여. 갱신 실행되면 업데이트 된 것을 복사합니다. pine_data_*.txt 트레이딩뷰의 파이프 편집기에 콘텐츠를 입력하고 클릭합니다 저장여러분의 차트 마커는 즉시 최신 발표 데이터를 반영합니다.

단계 6 FXMacroData 컨텍스트로 다시 TradingView 알림을 보내

지금까지의 작업 흐름은 매크로 데이터를 밀어냅니다. 로 트레이딩뷰. 단계 6는 다른 방향으로 루프를 닫습니다: 파이인 스크립트가 거래 신호를 발사 할 때 (예를 들어 CPI 인쇄 후 저항 이상으로 돌파), 트레이닝뷰의 웹 알림 시스템은 해당 이벤트를 작은 플래스크 엔드 포인트로 전송하여 즉시 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 Script 조건을 선택하고, 켜십시오 웹 룩 URL, 그리고 로 끝나는 ngrok HTTPS URL을 붙여넣어 /tradingview/alert. JSON 유타로드로 경고 메시지를 설정합니다:

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

파이인 스크립트 조건이 트리거될 때마다, 트레이딩뷰는 JSON을 엔드포인트에 게시합니다. 웹휴크 핸들러는 FXMacroData에서 최신 EUR 정책율과 CPI를 가져와 신호와 함께 전체 매크로 컨텍스트를 로그인합니다. 당신은 이것을 확장하여 풍부한 데이터를 사용하여 Telegram 또는 Slack 알림을 보낼 수 있습니다.

단계 7 COT 위치 및 FX 스팟 컨텍스트로 확장

매크로 발표 데이터는 그림의 한 층입니다. 더 완전한 신호를 위해서는 FXMacroData에서 CFTC 트레이더의 포지셔닝 약속 및 FX 스팟 트렌드 컨텍스트를 추가하십시오.

EUR 선물에 대한 Fetch 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
    }
  ]
}

단계 2와 같은 패턴을 사용하여 파이인 스크립트 생성기에 COT 네트워크 위치 시리즈를 추가합니다. net_noncommercial 이 지표는 유로 파워의 순환을 나타냅니다. COT 대시보드 표시기에 삽입하기 전에

파이썬 생성기를 확장하여 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 그래서 가격과 거시 발표와 함께 투기적 포지셔닝 트렌드를 볼 수 있습니다.