Forem: Robert Tidball

Introducing the FXMacroData Economic Data Embed Widget

Robert Tidball — Wed, 03 Dec 2025 13:34:36 +0000

Today, we’re launching a major new feature on FXMacroData — fully embeddable, real-time charts that you can drop into any website, blog, or research portal with a single copy-and-paste.

Each chart on FXMacroData now includes a dedicated “Embed Graph” button. Clicking it reveals a ready-to-paste HTML snippet.

You can view the full data and chart on the dashboard here: https://fxmacrodata.com/dashboard/EUR_USD To embed the chart above on your own site, use this snippet:

<!-- FXMacroData Chart Embed -->
<div
    id="fxmacro-carryChart"
    style="width: 100%; max-width: 800px; margin: 20px auto;"
>
    <iframe
        src="https://fxmacrodata.com/dashboard/embed/EUR_USD/carryChart"
        width="100%"
        height="450"
        frameborder="0"
        style="border: 1px solid #e5e7eb; border-radius: 8px;"
    ></iframe>
</div>

How Embeds Work

Embedding is as simple as copying the snippet and pasting it into your page. Each FXMacroData embed:

Updates in real time as new announcements are released
Requires zero maintenance or API integration
Loads quickly thanks to a lightweight hosting layer
Automatically adapts to your layout and device size
Built for Analysts, Traders, and Educators

Whether you’re building models, writing research, teaching, or running a fintech product, these embeds let you present live macroeconomic data without engineering work.

Designed for Reliability and Performance

Each widget runs on the same architecture powering the FXMacroData dashboards, backed by:

Real-time streaming data from our announcement pipeline
Optimized caching for low latency
Cloud Run auto-scaling for high-volatility events

Your embeds stay fast, accurate, and stable — even during major releases when traffic spikes.

Start Embedding Today

Visit any dashboard on FXMacroData and click “Embed Graph” to generate your snippet instantly.

Creating a python SDK for FXMacroData

Robert Tidball — Tue, 25 Nov 2025 13:46:19 +0000

🐍 Building an FX Trading Edge: Creating a Python Client for the FXMacroData API

Robert Tidball for FXMacroData ・ Nov 25

#python #api #github #programming

🐍 Building an FX Trading Edge: Creating a Python Client for the FXMacroData API

Robert Tidball — Tue, 25 Nov 2025 13:45:20 +0000

When building a Python library, the goal is to turn a complex, boilerplate-heavy process (raw API calls) into a simple, elegant one-liner. The FXMacroData API provides real-time macroeconomic indicators for major currency pairs—a goldmine for quant traders and analysts.

Raw API calls force developers to repeat code for authentication, error checking, and URL construction. Embracing the DRY (Don't Repeat Yourself) principle, I set out to build a dedicated Python library on top of it, creating a user-friendly wrapper. This article walks you through the core components of that wrapper, covering synchronous and asynchronous clients, proper exception handling, and utility functions.

1. Project Scaffolding and Handling Authentication

A good library starts with an intuitive entry point. My goal was to turn an HTTP request into a clean Python method call like:

client.get("aud", "inflation")

The Client Constructor

The Client class holds the base URL and API key. The FXMacroData API has a unique feature: USD data is public, but other currencies require an API key. The constructor handles this requirement upfront.

# client.py or async_client.py
from typing import Optional
from .exceptions import FXMacroDataError

class Client:
    BASE_URL = "https://fxmacrodata.com/api"

    def __init__(self, api_key: Optional[str] = None):
        """
        Synchronous FXMacroData Client.
        api_key: Required for non-USD currencies. USD is public.
        """
        self.api_key = api_key

2. Core Logic: The Synchronous Client (`Client`)

The synchronous Client uses the popular requests library. The main logic resides in the get method, which dynamically constructs the URL and enforces the API key requirement.

The `get` Method: Dynamic URL Construction and Key Check

# client.py
def get(
    self,
    currency: str,
    indicator: str,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
) -> dict:
    currency = currency.lower()
    url = f"{self.BASE_URL}/{currency}/{indicator}"

    headers = {}
    if currency != "usd":
        if not self.api_key:
            # Custom exception is crucial for user-friendly errors
            raise FXMacroDataError(f"API key required for {currency.upper()} endpoints.")
        headers["X-API-Key"] = self.api_key

    params = {}
    # ... params and API call logic ...

Robust Error Handling with Custom Exceptions

A robust library must handle failures gracefully. I created a custom exception, FXMacroDataError, to catch network issues and non-200 status codes, returning a clear, actionable message.

# exceptions.py
class FXMacroDataError(Exception):
    """Custom exception for FXMacroData client errors."""
    pass

Core request logic with the error wrapper:

# client.py (continued)
try:
    response = requests.get(url, headers=headers, params=params)
except Exception as e:
    raise FXMacroDataError(f"Request failed: {e}")

if response.status_code != 200:
    # Raise a clear error if the API returns a problem
    raise FXMacroDataError(f"API Error ({response.status_code}): {response.text}")

return response.json()

3. Advanced Feature: The Asynchronous Client (`AsyncClient`)

For automated trading bots or high-traffic dashboards, asynchronous programming is essential for performance. The AsyncClient uses the aiohttp library for non-blocking I/O.

Asynchronous Session Management

I implemented async context managers (__aenter__ and __aexit__) to ensure the aiohttp.ClientSession is created and properly closed, preventing resource leaks.

# async_client.py
import aiohttp
# ... imports ...

class AsyncClient:
    # ... init ...

    async def __aenter__(self) -> "AsyncClient":
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        if self.session:
            await self.session.close()
            self.session = None

This allows concurrent execution, where the total time is the maximum delay, not the sum:

import asyncio
from fxmacrodata import AsyncClient

async def main():
    async with AsyncClient(api_key="YOUR_KEY") as client:
        # Concurrent calls are now trivial
        data_aud = client.get("aud", "inflation")
        data_eur = client.get("eur", "gdp")

        aud_data, eur_data = await asyncio.gather(data_aud, data_eur)
        # ...

4. Utility: Cleaning Up the Data

Data consumers expect chronologically sorted data, but APIs don't always guarantee it. A small utility function ensures the output is always clean time-series data, checking for either a 'date' or 'release_date' key.

# utils.py
def sort_by_date(data_list):
    """Sorts a list of indicator data dictionaries by 'date' or 'release_date'."""
    return sorted(data_list, key=lambda x: x.get('date') or x.get('release_date'))

Building this wrapper solidified my understanding of Object-Oriented Design, the crucial performance trade-offs between Synchronous vs. Asynchronous networking, and the importance of a great Developer Experience through custom exceptions. You can explore the full source code on GitHub.

The final step was packaging the library and publishing it to PyPI. If you're building a tool to integrate real-time FX macro data, or just want a pattern for creating your own wrapper, the structure of this library is a solid foundation. Happy coding!

— Rob @ FXMacroData

How SuburbStory Uses Asyncio to Pull Multi-Source Government Data in Parallel

Robert Tidball — Mon, 24 Nov 2025 07:16:16 +0000

SuburbStory publishes local updates for every suburb in NSW. The data behind each update comes from several government and public feeds: police, fire, traffic, transport, and various agency alerts. Each of these sources is independent, updated on its own schedule, and exposed through completely different HTTP endpoints.

Because of that variety, the core engineering problem isn’t processing the data. It’s getting it fast enough, from all sources, without creating a slow and fragile chain of sequential API calls. The entire workload is I/O: pulling remote webpages, parsing lightweight text, and sending the processed information to Gemini to generate suburb-level summaries.

The only way to make this practical at state-wide scale is to run everything in parallel. That’s where Python’s asyncio becomes the centre of the system.

The Problem With Sequential Fetching

If you fetch each source one at a time, you create an artificial bottleneck. A single slow feed (for example a public incident feed that occasionally takes a few seconds to respond) blocks all other data from being processed. When you multiply that across hundreds of suburbs, you lose freshness and your cron job runtime becomes unpredictable.

But none of these tasks depend on each other. Each fetch is just an isolated HTTP call. There’s no reason to wait for one to finish before starting the next.

Using asyncio.gather to Hit Every Source at Once
The actual workflow is simple: as soon as the cron job starts, every data source is queried in parallel. Instead of a loop like:

for source in sources:
    data = fetch(source)
    process(data)

the pipeline builds a list of coroutines and launches them simultaneously:

results = await asyncio.gather(*tasks, return_exceptions=True)

Every request goes out at the same time. Each one resolves whenever the remote server responds. A slow feed no longer slows down anything else. A fast feed is processed immediately. When you have tens of government sources multiplied across thousands of suburbs, this difference is enormous.

Parsing and Grouping on the Fly

Once the fetches complete, the content is parsed and normalised. These steps are lightweight — mostly HTML extraction, field cleaning, text slicing, and location matching — so they run inline without blocking the loop. By the time all tasks in the gather call have resolved, you already have a full snapshot of every government update that matters for that cycle.

The next step is grouping by suburb. Because all sources arrive together, the system can assemble a complete picture of what happened in a suburb across police, fire, traffic, and other feeds in a single pass. That grouped data becomes the input to the NLG stage.

Parallel Model Requests to Gemini

Model calls are also I/O. They involve sending the suburb-level dataset to Gemini and waiting for a generated summary. Running these sequentially would be even slower than sequential scraping. But they fit the same pattern as the fetches: independent, high-latency network operations.

So the pipeline builds another batch of tasks (one model call per suburb) and fires them all at once using asyncio.gather, with concurrency limits where needed. While the model is working on one suburb, dozens of other tasks are in flight. No time is spent waiting for NLG work to complete before starting the next one.

This effectively compresses what would be minutes of serialized model calls down to the latency of a single batch.

The entire cycle looks like this:

All data sources fetched concurrently
All parsing done immediately after responses arrive
All suburb summaries generated concurrently
All output stored without blocking the loop

The cron job doesn’t wait on anything except the slowest item in each batch. Everything else is overlapped. Even though the system touches many external endpoints and calls an LLM at scale, the total runtime stays short and predictable because every part of the pipeline runs in parallel.

Why Asyncio Suits This Problem Exactly

The workload behind SuburbStory isn’t CPU-heavy. There’s no local ML model, no heavy processing, no long transformations. It’s almost entirely network-bound. Asyncio is designed for this exact shape of problem: many independent I/O operations that can safely run together.

Using threads or processes would add overhead and resource waste. Using synchronous requests would stretch each cron cycle unnecessarily. Asyncio sits neatly in the middle with minimal overhead, maximum concurrency, and predictable performance.

Flexible Flask Endpoints for Hierarchical Local News on SuburbStory

Robert Tidball — Mon, 17 Nov 2025 03:40:03 +0000

SuburbStory publishes automated local news pages for every suburb in NSW (Australia). The site’s URLs follow a clear geographic hierarchy:

https://suburbstory.com/<country>/<state>/<area>/<suburb>/

Each page is built by aggregating structured data from multiple government sources such as police, fire, traffic, and council feeds. This data is then processed via the Gemini API to generate readable content.

A key challenge is efficiently mapping thousands of geographic URL combinations to a single Flask application.

Dynamic Routing in Flask

Rather than creating a route for each suburb, SuburbStory uses a single, flexible Flask route:

@app.route("/<country>/<state>/<area>/<suburb>/")
def suburb_page(country, state, area, suburb):
    key = f"{country}/{state}/{area}/{suburb}"
    data = fetch_suburb_doc(key)  # fetch data from Firestore
    return render_template("suburb.html", data=data)

Flask automatically parses the URL segments and passes them as function arguments. The handler then retrieves the corresponding Firestore document and renders it with a single template. This allows one route to serve thousands of unique pages across different areas and suburbs without modifying the code.

Real Examples

Two suburbs under the same area illustrate the approach:

These pages use the same route and template while aggregating their own data, demonstrating how hierarchical Flask endpoints efficiently handle multiple suburbs.

Why Hierarchical URLs Matter

The hierarchical pattern country/state/area/suburb has several advantages:

Predictable and SEO-friendly URLs – easy to link externally and for search engines to index.
Direct mapping to Firestore – each page can be stored under a document key like "au/nsw/willoughby/chatswood", simplifying retrieval.
Template reusability – the same suburb.html template dynamically adjusts based on the URL variables.
Scalable architecture – Cloud Run handles all routes in one Flask app, allowing horizontal scaling without additional routing configuration.

Takeaways for Python Developers

SuburbStory demonstrates how flexible Flask routing enables:

Serving thousands of hierarchical pages from a single route
Efficiently handling independent data fetches per page
Scaling a serverless Flask app on Cloud Run without duplicating code
Maintaining clean, predictable, SEO-friendly URLs that map directly to Firestore

This is a practical model for building large-scale local or multi-tenant content systems in Python.

🚀 Scaling Up: Why I Chose FastAPI Over Flask and Django for a Data API

Robert Tidball — Thu, 13 Nov 2025 04:23:02 +0000

As a developer, when you set out to build a new data service—like my recent project, FXMacroData (check out the live API at FXMacroData - Real-time Forex Data) the framework choice is crucial. My goal was simple: serve high-frequency macroeconomic data instantly and reliably.

I narrowed the field down to the Python giants: Flask, Django, and FastAPI. While Flask is famously lightweight and Django is the enterprise powerhouse, I ultimately landed on FastAPI. Here's why that decision was a game-changer for building a performant, modern data API designed for the cloud.

The API Mandate: Speed, Concurrency, and Serverless

The core requirement for the FXMacroData API is high concurrency and an architecture optimized for modern cloud deployments, specifically serverless environments like Google Cloud Run.

Flask (Sync): Standard Flask is synchronous (WSGI), meaning it blocks a worker thread while waiting for I/O (like a database query). This inefficiency makes it harder to scale cost-effectively in a serverless environment where every millisecond of CPU time matters.
Django (Monolithic): Django is excellent, but it’s a heavyweight framework. For a pure API backend, its many built-in components were simply overkill. Deploying a massive framework just to serve a few data endpoints is inefficient, especially when using a flexible NoSQL backend like Firestore. Its complexity and synchronous default were hurdles to fast, cost-effective serverless deployment.

⚡️ FastAPI: Natively Async for Cloud Run

FastAPI is built on the modern ASGI standard, making it asynchronous (async/await) from the ground up. This was the single biggest performance advantage:

Non-Blocking I/O: Most API operations are I/O-bound (waiting for the database or network). Because FastAPI's worker doesn't block, it can efficiently handle hundreds of concurrent requests using minimal resources. This is essential for Cloud Run's scaling model.
Serverless Integration: Being lightweight and ASGI-native means FastAPI runs perfectly within the brief lifespan of a serverless container. It integrates cleanly with external services like Firestore, making it the ideal choice for a stateless, instant-scaling microservice.

The Code Difference: I/O Concurrency

The benefit of native asynchronous programming is immediately clear when requesting data from multiple sources.

➡️ Flask (Synchronous/Blocking)

The total execution time is the sum of the two delays (approx. 2 seconds), as the second call must wait for the first to complete.

# Flask (Synchronous)
import time
from flask import Flask
app = Flask(__name__)

# Execution runs sequentially: A waits for B to finish.
@app.route("/")
def sync_example():
    time.sleep(1)  # Wait for Source A
    time.sleep(1)  # Wait for Source B
    return "Total Time: ~2.0s"

➡️ FastAPI (Asynchronous/Non-Blocking)

The total execution time is the maximum of the two delays (approx. 1 second), as both I/O operations are initiated concurrently.

# FastAPI (Asynchronous)
import asyncio
from fastapi import FastAPI
app = FastAPI()

# Execution runs concurrently: A and B start at the same time.
@app.get("/")
async def async_example():
    await asyncio.gather(
        asyncio.sleep(1),  # Wait for Source A
        asyncio.sleep(1)   # Wait for Source B
    )
    return "Total Time: ~1.0s"

🧠 The Productivity Bonus: Clean Code and Auto-Docs

Beyond performance and cloud architecture, FastAPI delivered on developer experience:

It uses Pydantic models and Python type hints for automatic data validation and serialization.
It generates interactive OpenAPI documentation (Swagger UI) automatically.

Ultimately, FastAPI allowed me to build a high-performance, stateless API that perfectly matches the pay-per-use efficiency of Google Cloud Run, making it technically superior and far more cost-effective than using an over-engineered framework like Django for this specific task.

If you're building a new API focused on speed, data movement, and cloud-native scaling, the choice is clear: go async with FastAPI.

Forem: Robert Tidball

Introducing the FXMacroData Economic Data Embed Widget

How Embeds Work

Designed for Reliability and Performance

Start Embedding Today

Creating a python SDK for FXMacroData

🐍 Building an FX Trading Edge: Creating a Python Client for the FXMacroData API

Robert Tidball for FXMacroData ・ Nov 25

🐍 Building an FX Trading Edge: Creating a Python Client for the FXMacroData API

1. Project Scaffolding and Handling Authentication

The Client Constructor

2. Core Logic: The Synchronous Client (Client)

The get Method: Dynamic URL Construction and Key Check

Robust Error Handling with Custom Exceptions

3. Advanced Feature: The Asynchronous Client (AsyncClient)

Asynchronous Session Management

4. Utility: Cleaning Up the Data

How SuburbStory Uses Asyncio to Pull Multi-Source Government Data in Parallel

The Problem With Sequential Fetching

Parsing and Grouping on the Fly

Parallel Model Requests to Gemini

The entire cycle looks like this:

Why Asyncio Suits This Problem Exactly

Flexible Flask Endpoints for Hierarchical Local News on SuburbStory

Dynamic Routing in Flask

Real Examples

Why Hierarchical URLs Matter

Takeaways for Python Developers

🚀 Scaling Up: Why I Chose FastAPI Over Flask and Django for a Data API

The API Mandate: Speed, Concurrency, and Serverless

⚡️ FastAPI: Natively Async for Cloud Run

The Code Difference: I/O Concurrency

➡️ Flask (Synchronous/Blocking)

➡️ FastAPI (Asynchronous/Non-Blocking)

🧠 The Productivity Bonus: Clean Code and Auto-Docs

2. Core Logic: The Synchronous Client (`Client`)

The `get` Method: Dynamic URL Construction and Key Check

3. Advanced Feature: The Asynchronous Client (`AsyncClient`)