Forex API in Production: REST for Backtesting, WebSocket for Live Data

Most forex data problems in production don’t come from bad data. They come from using the right data through the wrong interface: a backtest that polls a live endpoint, a dashboard that refreshes by hammering REST every second, or a live strategy that silently trades on a feed that dropped for four minutes without anyone noticing.

The rule of thumb is short: use a REST API for backtesting and historical work, and a WebSocket API for real-time forex streaming. The useful part is everything around that rule: how to page through three years of minute bars without hitting rate limits, why your backtest needs spread data that historical candles don’t contain, and how to repair the hole a dropped connection leaves in your live data.

This guide walks through that architecture end to end, with Python examples against the Infoway Forex API. Each pattern carries over to other providers too.

What you’ll build:

• A rate-limit-aware historical backfill that pages backwards through 1-minute candles
• Validation checks that understand the forex weekly session (so weekends aren’t flagged as data loss)
• A spread model for backtests, built from live bid/ask data
• A resilient WebSocket consumer with heartbeats, backoff, and automatic gap repair over REST
• One canonical data model shared by the research and live paths

Why the Workload Decides the Interface

The two interfaces solve different problems:

A REST request is a question with a definite answer: “give me 500 one-minute EURUSD bars ending at this timestamp.” You can rerun it tomorrow, cache it, checkpoint it, and log exactly what you asked for. That’s what research needs.

A WebSocket connection is a subscription. The server pushes each update as it happens, without the latency and quota cost of polling. That’s what dashboards, alerts, signal engines, and risk monitors need.

Production systems need both, plus a defined way for the two to cover for each other.

One more distinction to settle up front: central-bank reference rates are not market data. The ECB’s daily euro reference rates are fine for accounting, invoicing, and month-end valuation, but they’re one fixing per day with no bid, no ask, and no intraday movement. Don’t backtest a trading strategy on them.

Part 1: REST for Backtesting

Know what the historical endpoint actually returns

Before writing any backfill code, pin down three things about your provider’s historical data, because each one changes your design:

1. Granularity. Is the history in ticks or in candles? Infoway’s REST history is candle-based: 1-minute up to yearly bars through POST /common/v2/batch_kline (the common path covers forex, metals, energy, and futures). Minute-level history goes back about three years; daily and higher intervals have no lookback limit.
2. Paging model. Infoway returns up to 500 bars per request for a single symbol, ending at the timestamp you pass (in seconds). To go further back, you move that timestamp backwards. Multi-symbol requests return only the latest 2 bars per symbol, so do historical backfills one pair at a time.
3. Ordering and types. Bars come back newest first, and prices are strings. Both will bite you if you assume otherwise.

Budget the backfill before you run it

Forex trades roughly 24 hours a day, 5 days a week, so one pair produces about 7,200 one-minute bars per week. Three years of 1-minute EURUSD is around 1.1 million bars, or about 2,250 requests at 500 bars each.

A 10-pair research universe on the free plan is a six-hour job, which is fine as an overnight run but not something to restart from zero every time a request fails. That’s why the backfill below is checkpointed and idempotent.

A backfill that pages backwards safely

import json
import os
import time
from decimal import Decimal
from pathlib import Path

import requests

API_KEY = os.environ["INFOWAY_API_KEY"]
BASE_URL = "https://data.infoway.io"
HEADERS = {"apiKey": API_KEY, "Content-Type": "application/json"}

REQUEST_INTERVAL = 1.0          # seconds between calls; match your plan's limit
RAW_DIR = Path("raw/klines")    # keep untouched responses for audit/replay


def fetch_klines(pair: str, kline_type: int, num: int, end_ts: int | None = None) -> list[dict]:
    """One REST call: up to `num` bars for one pair, ending at end_ts (seconds)."""
    body = {"klineType": kline_type, "klineNum": num, "codes": pair}
    if end_ts is not None:
        body["timestamp"] = end_ts

    for attempt in range(5):
        resp = requests.post(f"{BASE_URL}/common/v2/batch_kline",
                             json=body, headers=HEADERS, timeout=15)
        payload = resp.json() if resp.content else {}
        if resp.status_code == 429 or payload.get("ret") == 429:
            time.sleep(2 ** attempt)          # back off and retry
            continue
        resp.raise_for_status()
        if payload.get("ret") != 200:
            raise RuntimeError(f"{pair}: ret={payload.get('ret')} msg={payload.get('msg')} "
                               f"traceId={payload.get('traceId')}")

        # Persist the raw response before touching it
        RAW_DIR.mkdir(parents=True, exist_ok=True)
        (RAW_DIR / f"{pair}_{kline_type}_{end_ts}.json").write_text(json.dumps(payload))

        data = payload.get("data") or []
        return data[0]["respList"] if data else []
    raise RuntimeError(f"{pair}: still rate limited after retries")


def backfill(pair: str, start_ts: int, end_ts: int, kline_type: int = 1) -> list[dict]:
    """Walk backwards from end_ts to start_ts, 500 bars at a time."""
    bars: dict[int, dict] = {}                  # keyed by bar time -> dedupes overlaps
    cursor = end_ts

    while cursor > start_ts:
        batch = fetch_klines(pair, kline_type, 500, cursor)
        time.sleep(REQUEST_INTERVAL)
        if not batch:
            break

        for b in batch:
            t = int(b["t"])
            if start_ts <= t <= end_ts:
                bars[t] = {
                    "pair": pair,
                    "interval": "1m",
                    "ts": t,
                    "open": Decimal(b["o"]),
                    "high": Decimal(b["h"]),
                    "low": Decimal(b["l"]),
                    "close": Decimal(b["c"]),
                    "source": "rest",
                }

        oldest = min(int(b["t"]) for b in batch)
        if oldest >= cursor:                    # no progress: stop instead of looping forever
            break
        cursor = oldest - 1
        # save_checkpoint(pair, cursor)  -> resume from here after a crash

    return [bars[t] for t in sorted(bars)]      # API returns newest-first; store oldest-first

import json
import os
import time
from decimal import Decimal
from pathlib import Path

import requests

API_KEY = os.environ["INFOWAY_API_KEY"]
BASE_URL = "https://data.infoway.io"
HEADERS = {"apiKey": API_KEY, "Content-Type": "application/json"}

REQUEST_INTERVAL = 1.0          # seconds between calls; match your plan's limit
RAW_DIR = Path("raw/klines")    # keep untouched responses for audit/replay


def fetch_klines(pair: str, kline_type: int, num: int, end_ts: int | None = None) -> list[dict]:
    """One REST call: up to `num` bars for one pair, ending at end_ts (seconds)."""
    body = {"klineType": kline_type, "klineNum": num, "codes": pair}
    if end_ts is not None:
        body["timestamp"] = end_ts

    for attempt in range(5):
        resp = requests.post(f"{BASE_URL}/common/v2/batch_kline",
                             json=body, headers=HEADERS, timeout=15)
        payload = resp.json() if resp.content else {}
        if resp.status_code == 429 or payload.get("ret") == 429:
            time.sleep(2 ** attempt)          # back off and retry
            continue
        resp.raise_for_status()
        if payload.get("ret") != 200:
            raise RuntimeError(f"{pair}: ret={payload.get('ret')} msg={payload.get('msg')} "
                               f"traceId={payload.get('traceId')}")

        # Persist the raw response before touching it
        RAW_DIR.mkdir(parents=True, exist_ok=True)
        (RAW_DIR / f"{pair}_{kline_type}_{end_ts}.json").write_text(json.dumps(payload))

        data = payload.get("data") or []
        return data[0]["respList"] if data else []
    raise RuntimeError(f"{pair}: still rate limited after retries")


def backfill(pair: str, start_ts: int, end_ts: int, kline_type: int = 1) -> list[dict]:
    """Walk backwards from end_ts to start_ts, 500 bars at a time."""
    bars: dict[int, dict] = {}                  # keyed by bar time -> dedupes overlaps
    cursor = end_ts

    while cursor > start_ts:
        batch = fetch_klines(pair, kline_type, 500, cursor)
        time.sleep(REQUEST_INTERVAL)
        if not batch:
            break

        for b in batch:
            t = int(b["t"])
            if start_ts <= t <= end_ts:
                bars[t] = {
                    "pair": pair,
                    "interval": "1m",
                    "ts": t,
                    "open": Decimal(b["o"]),
                    "high": Decimal(b["h"]),
                    "low": Decimal(b["l"]),
                    "close": Decimal(b["c"]),
                    "source": "rest",
                }

        oldest = min(int(b["t"]) for b in batch)
        if oldest >= cursor:                    # no progress: stop instead of looping forever
            break
        cursor = oldest - 1
        # save_checkpoint(pair, cursor)  -> resume from here after a crash

    return [bars[t] for t in sorted(bars)]      # API returns newest-first; store oldest-first

A few decisions here are worth copying regardless of provider:

• Prices go into Decimal, not float. A float 1.08425 is not exactly 1.08425, and the error compounds across a million bars of P&L arithmetic.
• Raw responses are written to disk before parsing. When a backtest result looks wrong three weeks from now, you’ll want to know whether the problem is in the data or in your normalization code.
• Bars are keyed by timestamp. Overlapping pages, retried requests, and reruns produce the same result. This idempotency is what makes gap repair safe later.
• The oldest >= cursor guard stops an infinite loop if the API ever returns the same page twice.

Validate before research touches the data

Most forex validation code produces false alarms, because it treats the weekend as missing data. The market closes on Friday around 17:00 New York time and reopens on Sunday around 17:00 New York time (21:00 or 22:00 UTC depending on daylight saving). A validator has to know that.

from datetime import datetime, timezone


def fx_market_closed(ts: int) -> bool:
    """Conservative weekend window in UTC. Covers both DST regimes."""
    dt = datetime.fromtimestamp(ts, tz=timezone.utc)
    wd, hour = dt.weekday(), dt.hour             # Mon=0 ... Sun=6
    return (wd == 4 and hour >= 21) or wd == 5 or (wd == 6 and hour < 22)


def validate(bars: list[dict], step: int = 60) -> list[str]:
    issues = []
    for prev, cur in zip(bars, bars[1:]):
        if cur["ts"] <= prev["ts"]:
            issues.append(f"order/duplicate at {cur['ts']}")
        gap = range(prev["ts"] + step, cur["ts"], step)
        missing = [t for t in gap if not fx_market_closed(t)]
        if len(missing) > 5:                    # tolerate a few quiet minutes
            issues.append(f"gap of {len(missing)} open-market minutes after {prev['ts']}")

    for b in bars:
        if not (b["low"] <= min(b["open"], b["close"]) <= max(b["open"], b["close"]) <= b["high"]):
            issues.append(f"OHLC inconsistency at {b['ts']}")
        if b["low"] <= 0:
            issues.append(f"non-positive price at {b['ts']}")
    return issues

from datetime import datetime, timezone


def fx_market_closed(ts: int) -> bool:
    """Conservative weekend window in UTC. Covers both DST regimes."""
    dt = datetime.fromtimestamp(ts, tz=timezone.utc)
    wd, hour = dt.weekday(), dt.hour             # Mon=0 ... Sun=6
    return (wd == 4 and hour >= 21) or wd == 5 or (wd == 6 and hour < 22)


def validate(bars: list[dict], step: int = 60) -> list[str]:
    issues = []
    for prev, cur in zip(bars, bars[1:]):
        if cur["ts"] <= prev["ts"]:
            issues.append(f"order/duplicate at {cur['ts']}")
        gap = range(prev["ts"] + step, cur["ts"], step)
        missing = [t for t in gap if not fx_market_closed(t)]
        if len(missing) > 5:                    # tolerate a few quiet minutes
            issues.append(f"gap of {len(missing)} open-market minutes after {prev['ts']}")

    for b in bars:
        if not (b["low"] <= min(b["open"], b["close"]) <= max(b["open"], b["close"]) <= b["high"]):
            issues.append(f"OHLC inconsistency at {b['ts']}")
        if b["low"] <= 0:
            issues.append(f"non-positive price at {b['ts']}")
    return issues

Also check:

• The in-progress bar. The newest bar in a REST response is usually still forming. If your backfill ends at “now,” drop the last bar or overwrite it on the next run.
• Daily bar boundaries. Forex has no universal “day.” Providers close daily bars at UTC midnight, 17:00 New York, or something else. Rebuild a few daily bars from 1-minute data and compare before you use daily candles for anything session-sensitive.
• Holidays and thin sessions. Christmas and New Year’s produce real, legitimate gaps and wide spreads. Mark them; don’t interpolate them.

The backtesting pitfall that matters most: spread

Candles are built from a single price series. They don’t tell you what it would have cost to cross the spread at each moment. A backtest that buys and sells at the candle price is assuming a zero-cost market that doesn’t exist.

That’s not a rounding error. Take a strategy that makes 10 round trips a day on EURUSD with an average all-in cost of 0.8 pips per round trip. That’s 8 pips a day, or about 2,000 pips a year. Plenty of intraday strategies that look profitable on mid prices lose money once you subtract that.

Spread also isn’t constant. It’s tightest when London and New York overlap, widens through the Asian session for EUR and GBP pairs, and blows out around the 17:00 New York rollover and major data releases. A fixed spread assumption understates costs exactly where many strategies trade most.

The practical fix: record bid/ask live, build a spread profile, and apply it in the backtest.

from collections import defaultdict
from statistics import median

# Built from live depth snapshots (see Part 2): spread per pair, per hour-of-week
spread_samples: dict[tuple[str, int], list[Decimal]] = defaultdict(list)


def record_spread(pair: str, ts_ms: int, bid: Decimal, ask: Decimal) -> None:
    dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
    spread_samples[(pair, dt.weekday() * 24 + dt.hour)].append(ask - bid)


def expected_spread(pair: str, ts: int, fallback: Decimal) -> Decimal:
    dt = datetime.fromtimestamp(ts, tz=timezone.utc)
    samples = spread_samples.get((pair, dt.weekday() * 24 + dt.hour))
    return median(samples) if samples else fallback


def fill_price(side: str, ref_price: Decimal, spread: Decimal) -> Decimal:
    """Buys pay the ask, sells hit the bid."""
    half = spread / 2
    return ref_price + half if side == "buy" else ref_price - half

from collections import defaultdict
from statistics import median

# Built from live depth snapshots (see Part 2): spread per pair, per hour-of-week
spread_samples: dict[tuple[str, int], list[Decimal]] = defaultdict(list)


def record_spread(pair: str, ts_ms: int, bid: Decimal, ask: Decimal) -> None:
    dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
    spread_samples[(pair, dt.weekday() * 24 + dt.hour)].append(ask - bid)


def expected_spread(pair: str, ts: int, fallback: Decimal) -> Decimal:
    dt = datetime.fromtimestamp(ts, tz=timezone.utc)
    samples = spread_samples.get((pair, dt.weekday() * 24 + dt.hour))
    return median(samples) if samples else fallback


def fill_price(side: str, ref_price: Decimal, spread: Decimal) -> Decimal:
    """Buys pay the ask, sells hit the bid."""
    half = spread / 2
    return ref_price + half if side == "buy" else ref_price - half

Two more rules that prevent the most common self-inflicted backtest bugs:

• Never fill on the bar that generated the signal. If your signal uses a bar’s close, the earliest realistic fill is the next bar’s open. Filling at the same close is lookahead bias.
• Be careful when resampling. Building 15-minute bars from 1-minute bars is fine for prices, but the spread at the bar close is not the average spread over the bar. Keep spread on its own time series.

Part 2: WebSocket for Real-Time Forex Streaming

What to subscribe to

On Infoway, forex streams over:

wss://data.infoway.io/ws?business=common&apikey=...

with three subscription types:

Subscribing to both depth and 1-minute candles gives you the two things a live strategy needs: the same bar format your backtest used, and the real bid/ask you’ll actually trade against.

The rules that will disconnect you if you ignore them

• Heartbeat: send a 10010 message regularly. A connection with no heartbeat for 1 minute is closed (error 513). Every 30 seconds gives you margin.
• Request budget: subscribe, unsubscribe, and heartbeat messages share a limit of 60 per minute per connection. Subscribe to all pairs in one message (comma-separated codes), not one message per pair, or a reconnect with a large watchlist can trip the limit and get disconnected again in a loop.
• Connection count: plans allow a fixed number of connections (code 427 when exceeded). Make sure there’s exactly one reconnect loop. Duplicate reconnect tasks are a classic way to burn through it.

A resilient consumer with automatic gap repair

A reconnect restores the stream, not the data. Whatever happened while you were disconnected is gone from the WebSocket’s point of view, so the consumer must track the last bar it accepted and backfill the gap over REST after reconnecting.

import asyncio
import random
import uuid

import websockets

WS_URL = f"wss://data.infoway.io/ws?business=common&apikey={API_KEY}"
PAIRS = ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD"]


def msg(code: int, data: dict | None = None) -> str:
    m = {"code": code, "trace": uuid.uuid4().hex}
    if data is not None:
        m["data"] = data
    return json.dumps(m)


class ForexStream:
    def __init__(self, store):
        self.store = store                         # anything with upsert_bar() / write_quote()
        self.last_bar_ts: dict[str, int] = {}      # last *completed* 1-min bar per pair
        self.open_bar: dict[str, dict] = {}        # bar currently forming

    async def run(self) -> None:
        delay = 5
        while True:                                # the ONLY reconnect loop
            try:
                async with websockets.connect(WS_URL, ping_interval=None) as ws:
                    codes = ",".join(PAIRS)
                    await ws.send(msg(10003, {"codes": codes}))                         # depth
                    await ws.send(msg(10006, {"arr": [{"type": 1, "codes": codes}]}))  # 1-min bars
                    delay = 5
                    hb = asyncio.create_task(self._heartbeat(ws))
                    if self.last_bar_ts:                       # this is a reconnect
                        asyncio.create_task(self._repair_gaps())
                    try:
                        async for raw in ws:
                            self._dispatch(json.loads(raw))
                    finally:
                        hb.cancel()
            except (websockets.ConnectionClosed, OSError) as e:
                print(f"disconnected: {e!r}")
            await asyncio.sleep(delay + random.uniform(0, delay / 2))   # jittered backoff
            delay = min(delay * 2, 60)

    async def _heartbeat(self, ws) -> None:
        while True:
            await asyncio.sleep(30)
            await ws.send(msg(10010))

    def _dispatch(self, m: dict) -> None:
        code, data = m.get("code"), m.get("data")
        if code == 10005:                          # depth -> top of book
            bid, ask = Decimal(data["b"][0][0]), Decimal(data["a"][0][0])
            self.store.write_quote(normalize_quote(data["s"], bid, ask, data["t"], "websocket"))
            record_spread(data["s"], data["t"], bid, ask)
        elif code == 10008:                        # candle update
            self._on_bar(data)
        elif code not in (200, 10001, 10004, 10007, 10010, 10011):
            print(f"unexpected message: {m}")       # surface errors, don't swallow them

    def _on_bar(self, d: dict) -> None:
        pair, t = d["s"], int(d["t"])
        prev = self.open_bar.get(pair)
        # The current bar is pushed repeatedly as it updates.
        # It's final only once a bar with a later timestamp arrives.
        if prev and t > prev["ts"]:
            self.store.upsert_bar(prev)
            self.last_bar_ts[pair] = prev["ts"]
        self.open_bar[pair] = normalize_bar(pair, t, d, "websocket")

    async def _repair_gaps(self) -> None:
        now = int(time.time())
        for pair, last_ts in self.last_bar_ts.items():
            missing = (now - last_ts) // 60 + 2      # small overlap on purpose
            if missing <= 2:
                continue
            bars = await asyncio.to_thread(fetch_klines, pair, 1, min(missing, 500))
            for b in bars:
                t = int(b["t"])
                if last_ts < t < now // 60 * 60:     # completed bars only
                    self.store.upsert_bar(normalize_bar(pair, t, b, "rest"))
            await asyncio.sleep(REQUEST_INTERVAL)

import asyncio
import random
import uuid

import websockets

WS_URL = f"wss://data.infoway.io/ws?business=common&apikey={API_KEY}"
PAIRS = ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD"]


def msg(code: int, data: dict | None = None) -> str:
    m = {"code": code, "trace": uuid.uuid4().hex}
    if data is not None:
        m["data"] = data
    return json.dumps(m)


class ForexStream:
    def __init__(self, store):
        self.store = store                         # anything with upsert_bar() / write_quote()
        self.last_bar_ts: dict[str, int] = {}      # last *completed* 1-min bar per pair
        self.open_bar: dict[str, dict] = {}        # bar currently forming

    async def run(self) -> None:
        delay = 5
        while True:                                # the ONLY reconnect loop
            try:
                async with websockets.connect(WS_URL, ping_interval=None) as ws:
                    codes = ",".join(PAIRS)
                    await ws.send(msg(10003, {"codes": codes}))                         # depth
                    await ws.send(msg(10006, {"arr": [{"type": 1, "codes": codes}]}))  # 1-min bars
                    delay = 5
                    hb = asyncio.create_task(self._heartbeat(ws))
                    if self.last_bar_ts:                       # this is a reconnect
                        asyncio.create_task(self._repair_gaps())
                    try:
                        async for raw in ws:
                            self._dispatch(json.loads(raw))
                    finally:
                        hb.cancel()
            except (websockets.ConnectionClosed, OSError) as e:
                print(f"disconnected: {e!r}")
            await asyncio.sleep(delay + random.uniform(0, delay / 2))   # jittered backoff
            delay = min(delay * 2, 60)

    async def _heartbeat(self, ws) -> None:
        while True:
            await asyncio.sleep(30)
            await ws.send(msg(10010))

    def _dispatch(self, m: dict) -> None:
        code, data = m.get("code"), m.get("data")
        if code == 10005:                          # depth -> top of book
            bid, ask = Decimal(data["b"][0][0]), Decimal(data["a"][0][0])
            self.store.write_quote(normalize_quote(data["s"], bid, ask, data["t"], "websocket"))
            record_spread(data["s"], data["t"], bid, ask)
        elif code == 10008:                        # candle update
            self._on_bar(data)
        elif code not in (200, 10001, 10004, 10007, 10010, 10011):
            print(f"unexpected message: {m}")       # surface errors, don't swallow them

    def _on_bar(self, d: dict) -> None:
        pair, t = d["s"], int(d["t"])
        prev = self.open_bar.get(pair)
        # The current bar is pushed repeatedly as it updates.
        # It's final only once a bar with a later timestamp arrives.
        if prev and t > prev["ts"]:
            self.store.upsert_bar(prev)
            self.last_bar_ts[pair] = prev["ts"]
        self.open_bar[pair] = normalize_bar(pair, t, d, "websocket")

    async def _repair_gaps(self) -> None:
        now = int(time.time())
        for pair, last_ts in self.last_bar_ts.items():
            missing = (now - last_ts) // 60 + 2      # small overlap on purpose
            if missing <= 2:
                continue
            bars = await asyncio.to_thread(fetch_klines, pair, 1, min(missing, 500))
            for b in bars:
                t = int(b["t"])
                if last_ts < t < now // 60 * 60:     # completed bars only
                    self.store.upsert_bar(normalize_bar(pair, t, b, "rest"))
            await asyncio.sleep(REQUEST_INTERVAL)

Why this works:

1. Reconnect first, repair second. The stream comes back before the REST backfill starts, so no second gap opens while you’re repairing the first.
2. The overlap is deliberate. Repair requests a couple of extra bars, and because bars are upserted by (pair, interval, ts), overlapping REST and WebSocket bars collapse into one record. Without idempotent writes, every repair creates duplicates.
3. Only completed bars are written. The forming bar is left to the live stream.

Be clear about what can and can’t be recovered. Missing 1-minute bars can be fully restored over REST. Missing bid/ask snapshots can’t, because there’s no historical depth endpoint to replay them from. Log those windows as known gaps in the quote series rather than pretending they’re complete. For long outages (more than 500 minutes), reuse the paged backfill() from Part 1 instead of a single request.

One Canonical Data Model for Both Paths

Your strategy code should never know whether a bar came from REST or WebSocket. The two sources disagree on almost everything at the format level:

So everything passes through one normalization layer:

def normalize_bar(pair: str, ts: int, raw: dict, source: str) -> dict:
    return {
        "pair": pair,                       # symbol, e.g. "EURUSD"
        "base": pair[:3], "quote": pair[3:],
        "interval": "1m",
        "ts": int(ts),                      # bar open time, UTC seconds
        "open": Decimal(raw["o"]), "high": Decimal(raw["h"]),
        "low": Decimal(raw["l"]),  "close": Decimal(raw["c"]),
        "source": source,                   # "rest" | "websocket"
        "received_at": time.time(),
    }


def normalize_quote(pair: str, bid: Decimal, ask: Decimal, ts_ms: int, source: str) -> dict:
    return {
        "pair": pair,
        "base": pair[:3], "quote": pair[3:],
        "bid": bid, "ask": ask,
        "mid": (bid + ask) / 2,
        "spread": ask - bid,
        "event_ts_ms": int(ts_ms),          # provider event time
        "received_ts_ms": int(time.time() * 1000),
        "source": source,
    }

def normalize_bar(pair: str, ts: int, raw: dict, source: str) -> dict:
    return {
        "pair": pair,                       # symbol, e.g. "EURUSD"
        "base": pair[:3], "quote": pair[3:],
        "interval": "1m",
        "ts": int(ts),                      # bar open time, UTC seconds
        "open": Decimal(raw["o"]), "high": Decimal(raw["h"]),
        "low": Decimal(raw["l"]),  "close": Decimal(raw["c"]),
        "source": source,                   # "rest" | "websocket"
        "received_at": time.time(),
    }


def normalize_quote(pair: str, bid: Decimal, ask: Decimal, ts_ms: int, source: str) -> dict:
    return {
        "pair": pair,
        "base": pair[:3], "quote": pair[3:],
        "bid": bid, "ask": ask,
        "mid": (bid + ask) / 2,
        "spread": ask - bid,
        "event_ts_ms": int(ts_ms),          # provider event time
        "received_ts_ms": int(time.time() * 1000),
        "source": source,
    }

Keeping both the event time and the receive time costs almost nothing and pays for itself the first time you investigate latency. The difference between the two is your end-to-end lag, and a growing gap between them is often the earliest sign that a consumer is falling behind.

Reference Architecture

Monitor these, per pair rather than just per connection:

• Freshness: seconds since the last quote. During market hours, a pair that goes quiet while others keep updating means a broken subscription, not a quiet market.
• Reconnect count and repair volume: how often you reconnect, and how many bars each repair writes.
• Lag: received_ts - event_ts, at the 50th and 99th percentile.
• Validation failures on each backfill run.
• REST 429s: if they show up regularly, your request interval is wrong.

A One-Day Proof of Concept

Before committing to any forex data provider, run this test. It surfaces most integration problems in a day:

1. Backfill one month of 1-minute EURUSD over REST. Run the validator. Expect zero open-market gaps larger than a few minutes.
2. Stream the same pair over WebSocket for a few hours, recording depth and 1-minute bars.
3. Compare overlapping bars. Bars from REST and WebSocket for the same minutes should match after normalization. If they don’t, find out why before you build anything else.
4. Kill the connection (pull the network, or kill the process) for 10 minutes during market hours. Restart. Confirm the repair job restores every missing bar and writes no duplicates.
5. Inspect the spread profile. Spreads should clearly widen around 17:00 New York time and tighten during the London/New York overlap. If spread looks flat all day, you’re not looking at real bid/ask data.

Pass criteria: no unexplained gaps, identical schemas from both paths, repair completes within a minute of reconnecting, and duplicate count after repair is zero.

Things to verify with any provider while you’re at it: pair coverage, historical depth per interval, whether bid/ask is available (live and historical), timestamp semantics, the daily bar boundary, rate and connection limits, and whether commercial use is covered by your plan.

Which Interface for Which Job

The architecture comes down to three rules. Use REST for anything bounded and repeatable. Use WebSocket for anything continuous. Connect the two with a shared data model and a gap-repair path you have actually tested by pulling the plug.

Try It With Infoway API

Infoway API provides forex, precious metals, energy, stocks, and crypto through the same REST and WebSocket interfaces, so the architecture above extends to multi-asset systems without a second integration. New accounts get a 7-day free trial with no credit card required, which is enough time to run the proof of concept end to end.

• Create a free account
• API documentation