Python Examples

# forex_client.py
import requests
import time
from typing import Optional


class ForexClient:
    BASE_URL = "https://api.pipdata.net"

    def __init__(self, api_key: str):
        self.session = requests.Session()
        self.session.headers.update({"X-API-Key": api_key})

    def _get(self, path: str, params: dict = None) -> dict:
        resp = self.session.get(f"{self.BASE_URL}{path}", params=params, timeout=15)
        resp.raise_for_status()
        return resp.json()

    def symbols(self, search: Optional[str] = None, limit: int = 100) -> list:
        """List available symbols, optionally filtered by name."""
        params = {"limit": limit}
        if search:
            params["search"] = search
        return self._get("/v1/symbols", params)["symbols"]

    def history(self, symbol: str, resolution: str, countback: int = 500) -> dict:
        """Fetch the latest N OHLCV bars for a symbol."""
        return self._get("/v1/history", {
            "symbol":     symbol,
            "resolution": resolution,
            "countback":  countback
        })

    def history_range(self, symbol: str, resolution: str, from_ts: int, to_ts: int) -> dict:
        """Fetch OHLCV bars between two Unix timestamps."""
        return self._get("/v1/history", {
            "symbol":     symbol,
            "resolution": resolution,
            "from":       from_ts,
            "to":         to_ts
        })

    def coverage(self, symbol: Optional[str] = None) -> dict:
        """Get available data date ranges."""
        params = {"symbol": symbol} if symbol else {}
        return self._get("/v1/coverage", params)

    def status(self) -> dict:
        """Check feed health status."""
        return self._get("/v1/status")

    def usage(self, days: int = 7) -> dict:
        """Get API key usage statistics."""
        return self._get("/v1/usage", {"days": days})


# Quick test
if __name__ == "__main__":
    client = ForexClient("fxd_your_key_here")

    st = client.status()
    print(f"Service: {st['status']}")

    data = client.history("WMMARKETS:ZERO:EURUSD", "60", countback=5)
    if data["s"] == "ok":
        print(f"EUR/USD last close: {data['c'][-1]:.5f}")

import pandas as pd
import numpy as np


def to_dataframe(history: dict) -> pd.DataFrame:
    """Convert API response to a timezone-aware pandas DataFrame."""
    if history.get("s") != "ok":
        return pd.DataFrame()

    df = pd.DataFrame({
        "time":   pd.to_datetime(history["t"], unit="s", utc=True),
        "open":   history["o"],
        "high":   history["h"],
        "low":    history["l"],
        "close":  history["c"],
        "volume": history["v"],
    })
    df.set_index("time", inplace=True)
    return df


def add_ema(df: pd.DataFrame, period: int) -> pd.DataFrame:
    df[f"ema{period}"] = df["close"].ewm(span=period, adjust=False).mean()
    return df


def add_rsi(df: pd.DataFrame, period: int = 14) -> pd.DataFrame:
    delta = df["close"].diff()
    gain  = delta.clip(lower=0).rolling(period).mean()
    loss  = (-delta.clip(upper=0)).rolling(period).mean()
    rs    = gain / loss
    df["rsi"] = 100 - (100 / (1 + rs))
    return df


def add_atr(df: pd.DataFrame, period: int = 14) -> pd.DataFrame:
    hl  = df["high"] - df["low"]
    hpc = (df["high"] - df["close"].shift()).abs()
    lpc = (df["low"]  - df["close"].shift()).abs()
    tr  = pd.concat([hl, hpc, lpc], axis=1).max(axis=1)
    df["atr"] = tr.ewm(span=period, adjust=False).mean()
    return df


# Example
client = ForexClient("fxd_your_key_here")
raw = client.history("WMMARKETS:ZERO:EURUSD", "60", 500)
df  = to_dataframe(raw)
df  = add_ema(df, 20)
df  = add_ema(df, 50)
df  = add_rsi(df, 14)
df  = add_atr(df, 14)

print(df[["close", "ema20", "ema50", "rsi", "atr"]].tail(5))

import csv
import os
import time


def download_all(client: ForexClient, resolution: str = "60", countback: int = 500):
    """
    Download OHLCV history for every accessible symbol and save to individual CSV files.
    """
    os.makedirs("forex_data", exist_ok=True)

    symbols = client.symbols(limit=1000)
    print(f"Downloading {len(symbols)} symbols at {resolution} resolution...")

    for i, sym in enumerate(symbols, 1):
        ticker = sym["symbol"]
        print(f"[{i:3}/{len(symbols)}] {ticker}...", end=" ", flush=True)

        data = client.history(ticker, resolution, countback)

        if data.get("s") != "ok":
            print("no data")
            continue

        fname = f"forex_data/{ticker.replace(':', '_')}_{resolution}.csv"
        with open(fname, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["time", "open", "high", "low", "close", "volume"])
            for row in zip(data["t"], data["o"], data["h"], data["l"], data["c"], data["v"]):
                writer.writerow(row)

        print(f"{len(data['t'])} bars saved")
        time.sleep(0.05)  # ~20 req/s, well within rate limits

    print("\nDone.")


client = ForexClient("fxd_your_key_here")
download_all(client, resolution="60", countback=500)

# live_stream.py
import asyncio
import json
import websockets
from datetime import datetime, timezone

API_KEY = "fxd_your_key_here"
URI     = f"wss://api.pipdata.net/v1/stream?api_key={API_KEY}"

SYMBOLS = [
    "WMMARKETS:ZERO:EURUSD",
    "WMMARKETS:ZERO:GBPUSD",
    "WMMARKETS:ZERO:XAUUSD",
    "WMMARKETS:ZERO:BTCUSD",
]


async def on_bar(symbol: str, bar: dict):
    """Process a newly closed M1 bar."""
    t   = datetime.fromtimestamp(bar["time"], tz=timezone.utc)
    sym = symbol.split(":")[-1]
    print(
        f"[{t.strftime('%H:%M')}] {sym:<8} "
        f"O={bar['open']:.5f} H={bar['high']:.5f} "
        f"L={bar['low']:.5f} C={bar['close']:.5f}"
    )


async def stream():
    reconnect_delay = 5

    while True:
        try:
            async with websockets.connect(URI, ping_interval=30) as ws:
                print("Connected ✓")
                reconnect_delay = 5

                await ws.send(json.dumps({
                    "action":  "subscribe",
                    "symbols": SYMBOLS,
                    "type":    "bar"
                }))

                async for raw in ws:
                    msg = json.loads(raw)

                    if msg["type"] == "connected":
                        print(f"Session: {msg['session_id']} | Max symbols: {msg['limits']['max_symbols']}")

                    elif msg["type"] == "subscribed":
                        print(f"Subscribed: {len(msg['symbols'])} symbols")
                        if msg.get("rejected"):
                            print(f"Rejected: {msg['rejected']}")

                    elif msg["type"] == "bar":
                        await on_bar(msg["symbol"], msg["bar"])

                    elif msg["type"] == "error":
                        print(f"Error: {msg['code']}")
                        if msg["code"] in ("KEY_INVALID", "KEY_INACTIVE", "KEY_EXPIRED"):
                            return  # fatal auth error — stop

        except websockets.exceptions.ConnectionClosed as e:
            print(f"Disconnected ({e}) — reconnecting in {reconnect_delay}s")
            await asyncio.sleep(reconnect_delay)
            reconnect_delay = min(reconnect_delay * 2, 60)

        except Exception as e:
            print(f"Unexpected error: {e}")
            await asyncio.sleep(reconnect_delay)


if __name__ == "__main__":
    asyncio.run(stream())

def latest_prices(client: ForexClient) -> dict:
    """Return the latest close price for all accessible symbols."""
    symbols = client.symbols(limit=1000)
    prices  = {}

    for sym in symbols:
        ticker = sym["symbol"]
        data   = client.history(ticker, "1", countback=1)
        if data.get("s") == "ok" and data["c"]:
            prices[ticker] = {
                "display": sym["display_symbol"],
                "close":   data["c"][-1],
                "digits":  sym["digits"]
            }

    return prices


client = ForexClient("fxd_your_key_here")
prices = latest_prices(client)

for ticker, info in list(prices.items())[:10]:
    fmt = f"{{:.{info['digits']}f}}"
    print(f"{info['display']:<12} = {fmt.format(info['close'])}")