Macro Data Release Reaction Engine: CPI, NFP, and FOMC in Under 60 Seconds

                  Scheduled trigger (release time, ET)
                                 │
                                 ▼
                         [ReleaseFetcher]
                  (BLS / FRED / consensus / history)
                                 │
                                 ▼
              [HypothesisReasoner — Opus 4.8 high]
            (surprise vs. consensus, regime hypothesis)
                                 │
                ┌────────────────┼────────────────┐
                ▼                ▼                ▼                ▼
        [RatesStrategist] [FXStrategist]  [EquityStrategist] [CommodityStrategist]
            gpt-4.1            grok-4         sonnet-4.5         gpt-4.1-mini
                └────────────────┼────────────────┘
                                 ▼
                  [ThesisSynthesizer — sonnet-4.5]
                                 │
                                 ▼
                         post_slack_alert()

pip install requests python-dotenv pytz
export SWARMS_API_KEY="your-premium-api-key"
export BLS_API_KEY="your-bls-api-key"
export FRED_API_KEY="your-fred-api-key"
export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/..."

import json
import os
import time
from datetime import datetime, timedelta

import requests
import pytz
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("SWARMS_API_KEY")
BASE_URL = "https://api.swarms.world"
ET = pytz.timezone("America/New_York")

headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}

FETCH_BLS_CPI = {
    "type": "function",
    "function": {
        "name": "fetch_bls_cpi",
        "description": (
            "Fetch the latest CPI release from BLS. Returns headline MoM, "
            "core MoM, headline YoY, core YoY, and the shelter and "
            "services-ex-shelter components."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "release_month": {
                    "type": "string",
                    "description": "Release month in YYYY-MM format.",
                },
            },
            "required": ["release_month"],
        },
    },
}

FETCH_BLS_NFP = {
    "type": "function",
    "function": {
        "name": "fetch_bls_nfp",
        "description": (
            "Fetch the latest Nonfarm Payrolls release from BLS. Returns "
            "headline payrolls change, unemployment rate, labor force "
            "participation, average hourly earnings MoM and YoY, and prior "
            "two months' revisions."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "release_month": {
                    "type": "string",
                    "description": "Release month in YYYY-MM format.",
                },
            },
            "required": ["release_month"],
        },
    },
}

FETCH_FRED_SERIES = {
    "type": "function",
    "function": {
        "name": "fetch_fred_series",
        "description": (
            "Fetch any FRED time series by series_id (e.g., DGS10, DFF, "
            "DEXUSEU, VIXCLS). Returns the most recent value and the "
            "trailing 90-day window for regime context."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "series_id": {
                    "type": "string",
                    "description": "FRED series identifier.",
                },
            },
            "required": ["series_id"],
        },
    },
}

LOOKUP_CONSENSUS = {
    "type": "function",
    "function": {
        "name": "lookup_consensus",
        "description": (
            "Look up the Wall Street consensus expectation for a given "
            "macro release (Bloomberg/Reuters survey median, mean, and "
            "high-low range)."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "release_id": {
                    "type": "string",
                    "description": "Release identifier (CPI, NFP, FOMC, GDP, PCE).",
                },
                "release_date": {
                    "type": "string",
                    "description": "Release date in YYYY-MM-DD format.",
                },
            },
            "required": ["release_id", "release_date"],
        },
    },
}

LOOKUP_HISTORICAL_REACTION = {
    "type": "function",
    "function": {
        "name": "lookup_historical_reaction",
        "description": (
            "Look up the empirical asset-class reaction in the 60 minutes "
            "post-release for comparable historical surprises in the same "
            "regime. Returns medians and 25/75 percentiles for 2y, 10y, "
            "DXY, SPX, gold, and crude moves."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "release_id": {"type": "string"},
                "surprise_zscore": {
                    "type": "number",
                    "description": "Standardized surprise (actual - consensus) / historical_std.",
                },
                "regime": {
                    "type": "string",
                    "description": "Current regime tag (e.g., 'hiking', 'cutting', 'pause', 'risk-on', 'risk-off').",
                },
            },
            "required": ["release_id", "surprise_zscore", "regime"],
        },
    },
}

FETCH_MARKET_SNAPSHOT = {
    "type": "function",
    "function": {
        "name": "fetch_market_snapshot",
        "description": (
            "Fetch a live pre-release market snapshot for an asset class. "
            "Returns levels, 1-day and 1-week changes, implied vol, and "
            "positioning indicators."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "asset_class": {
                    "type": "string",
                    "enum": ["rates", "fx", "equities", "commodities"],
                },
            },
            "required": ["asset_class"],
        },
    },
}

POST_SLACK_ALERT = {
    "type": "function",
    "function": {
        "name": "post_slack_alert",
        "description": (
            "Post the synthesized trade brief to a Slack channel via "
            "incoming webhook. Use markdown formatting."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "channel": {
                    "type": "string",
                    "description": "Slack channel (e.g., '#macro-desk').",
                },
                "text": {
                    "type": "string",
                    "description": "The formatted trade brief.",
                },
            },
            "required": ["channel", "text"],
        },
    },
}

ALL_TOOLS = [
    FETCH_BLS_CPI,
    FETCH_BLS_NFP,
    FETCH_FRED_SERIES,
    LOOKUP_CONSENSUS,
    LOOKUP_HISTORICAL_REACTION,
    FETCH_MARKET_SNAPSHOT,
    POST_SLACK_ALERT,
]

HYPOTHESIS_PROMPT = (
    "You are the senior macro strategist running the post-release "
    "hypothesis desk. Given a fresh macro print, you must produce a "
    "single regime-tagged hypothesis in under 15 seconds. Process:\n\n"
    "1. Call the appropriate fetcher (fetch_bls_cpi, fetch_bls_nfp, or "
    "   fetch_fred_series) to obtain the actual print.\n"
    "2. Call lookup_consensus to obtain Wall Street consensus.\n"
    "3. Compute the standardized surprise (actual - consensus) / std for "
    "   every relevant sub-component.\n"
    "4. Call fetch_fred_series for the current rate regime context (DFF, "
    "   DGS2, DGS10) and risk regime (VIXCLS, DXY).\n"
    "5. Call lookup_historical_reaction with your computed surprise and "
    "   regime tag.\n"
    "6. Emit a single JSON object with fields: surprise_summary, "
    "   regime_tag, primary_hypothesis (one sentence), and "
    "   confidence ('low'|'medium'|'high').\n\n"
    "Do NOT recommend trades — that is the job of the asset-class desks. "
    "Your output is the regime hypothesis they branch on. Be surgical."
)

HYPOTHESIS_AGENT = {
    "agent_name": "HypothesisReasoner",
    "description": "Computes the surprise and regime hypothesis.",
    "system_prompt": HYPOTHESIS_PROMPT,
    "model_name": "claude-opus-4-8",
    "reasoning_effort": "high",
    "max_loops": 1,
    "max_tokens": 4000,
    "temperature": 0.2,
    "tools_dictionary": [
        FETCH_BLS_CPI,
        FETCH_BLS_NFP,
        FETCH_FRED_SERIES,
        LOOKUP_CONSENSUS,
        LOOKUP_HISTORICAL_REACTION,
    ],
}

RATES_PROMPT = (
    "You are the rates desk strategist. Given the regime hypothesis, "
    "call fetch_market_snapshot('rates') and produce a single trade "
    "thesis in under 5 seconds: front-end vs. back-end positioning, "
    "specific instrument (2y, 5y, 10y futures or swaps), direction, "
    "size guidance, stop, and time horizon. Max 80 words. Be specific."
)

FX_PROMPT = (
    "You are the FX desk strategist. Given the regime hypothesis, call "
    "fetch_market_snapshot('fx') and produce a single trade thesis in "
    "under 5 seconds: which pair, direction, entry level, stop, size, "
    "and the carry/momentum lens that drives the call. Max 80 words. "
    "Be specific."
)

EQUITY_PROMPT = (
    "You are the equity desk strategist. Given the regime hypothesis, "
    "call fetch_market_snapshot('equities') and produce a single trade "
    "thesis in under 5 seconds: sector rotation call, specific "
    "instrument (futures, ETF, or pair), direction, stop, and how the "
    "regime shift cascades through cyclicals vs. defensives. Max 80 "
    "words. Be specific."
)

COMMODITY_PROMPT = (
    "You are the commodity desk strategist. Given the regime hypothesis, "
    "call fetch_market_snapshot('commodities') and produce a single "
    "trade thesis in under 5 seconds: gold or crude or both, direction, "
    "specific instrument, stop, and the dollar/real-rate transmission "
    "logic. Max 80 words. Be specific."
)

SYNTH_PROMPT = (
    "You are the chief macro strategist. Given the regime hypothesis "
    "and the four desk theses, produce a single Slack-ready trade "
    "brief in this exact format:\n\n"
    "*REGIME:* <tag>\n"
    "*HYPOTHESIS:* <one sentence>\n"
    "*RATES:* <thesis>\n"
    "*FX:* <thesis>\n"
    "*EQUITIES:* <thesis>\n"
    "*COMMODITIES:* <thesis>\n"
    "*CONVICTION:* <LOW|MEDIUM|HIGH>\n\n"
    "Then call post_slack_alert with channel='#macro-desk' and the "
    "brief as text. Do not editorialize across desks — preserve each "
    "thesis verbatim."
)


def build_release_swarm(release_id: str, release_ts: str) -> dict:
    return {
        "name": f"Macro-Release-Reaction-{release_id}-{release_ts}",
        "description": (
            "Scheduled reaction engine. Reasoning hypothesis fans out "
            "to four asset-class desks, then a synthesizer posts to Slack."
        ),
        "swarm_type": "GraphWorkflow",
        "max_loops": 1,
        "task": (
            f"Macro release {release_id} crossed at {release_ts} ET. "
            f"Run the full reaction pipeline: fetch the print, compute "
            f"the regime hypothesis, branch into asset desks, synthesize, "
            f"and post the brief to Slack."
        ),
        "agents": [
            HYPOTHESIS_AGENT,
            {
                "agent_name": "RatesStrategist",
                "description": "Rates desk thesis.",
                "system_prompt": RATES_PROMPT,
                "model_name": "gpt-4.1",
                "max_loops": 1,
                "max_tokens": 1500,
                "temperature": 0.3,
                "tools_dictionary": [FETCH_MARKET_SNAPSHOT],
            },
            {
                "agent_name": "FXStrategist",
                "description": "FX desk thesis.",
                "system_prompt": FX_PROMPT,
                "model_name": "grok-4",
                "max_loops": 1,
                "max_tokens": 1500,
                "temperature": 0.3,
                "tools_dictionary": [FETCH_MARKET_SNAPSHOT],
            },
            {
                "agent_name": "EquityStrategist",
                "description": "Equity desk thesis.",
                "system_prompt": EQUITY_PROMPT,
                "model_name": "claude-sonnet-4.5",
                "max_loops": 1,
                "max_tokens": 1500,
                "temperature": 0.3,
                "tools_dictionary": [FETCH_MARKET_SNAPSHOT],
            },
            {
                "agent_name": "CommodityStrategist",
                "description": "Commodities desk thesis.",
                "system_prompt": COMMODITY_PROMPT,
                "model_name": "gpt-4.1-mini",
                "max_loops": 1,
                "max_tokens": 1500,
                "temperature": 0.3,
                "tools_dictionary": [FETCH_MARKET_SNAPSHOT],
            },
            {
                "agent_name": "ThesisSynthesizer",
                "description": "Synthesizes the four desks into one Slack brief.",
                "system_prompt": SYNTH_PROMPT,
                "model_name": "claude-sonnet-4.5",
                "max_loops": 1,
                "max_tokens": 2000,
                "temperature": 0.2,
                "tools_dictionary": [POST_SLACK_ALERT],
            },
        ],
        "edges": [
            {"source": "HypothesisReasoner", "target": "RatesStrategist"},
            {"source": "HypothesisReasoner", "target": "FXStrategist"},
            {"source": "HypothesisReasoner", "target": "EquityStrategist"},
            {"source": "HypothesisReasoner", "target": "CommodityStrategist"},
            {"source": "RatesStrategist", "target": "ThesisSynthesizer"},
            {"source": "FXStrategist", "target": "ThesisSynthesizer"},
            {"source": "EquityStrategist", "target": "ThesisSynthesizer"},
            {"source": "CommodityStrategist", "target": "ThesisSynthesizer"},
        ],
        "entry_points": ["HypothesisReasoner"],
        "end_points": ["ThesisSynthesizer"],
        "auto_compile": True,
    }


def fire_release_swarm(release_id: str, release_ts: str) -> dict:
    payload = build_release_swarm(release_id, release_ts)
    response = requests.post(
        f"{BASE_URL}/v1/swarm/completions",
        headers=headers,
        json=payload,
        timeout=90,
    )
    response.raise_for_status()
    return response.json()

RELEASE_CALENDAR = [
    {"release_id": "CPI", "ts_et": "2026-06-12T08:30:00"},
    {"release_id": "NFP", "ts_et": "2026-06-06T08:30:00"},
    {"release_id": "FOMC", "ts_et": "2026-06-18T14:00:00"},
    # ... ~50 events per year
]


def schedule_release(release_id: str, ts_et_str: str) -> None:
    """Sleep until the release minute, then fire immediately."""
    release_ts = ET.localize(datetime.fromisoformat(ts_et_str))
    now_et = datetime.now(ET)
    wait_seconds = (release_ts - now_et).total_seconds()

    if wait_seconds < 0:
        print(f"Skipping {release_id} — already crossed.")
        return

    # Wake up 2 seconds before the release to be hot
    sleep_until = wait_seconds - 2
    if sleep_until > 0:
        print(f"Sleeping {sleep_until:.0f}s until {release_id} at {ts_et_str} ET")
        time.sleep(sleep_until)

    # Poll tight until the release timestamp crosses
    while datetime.now(ET) < release_ts:
        time.sleep(0.1)

    print(f"FIRE {release_id} at {datetime.now(ET).isoformat()}")
    t0 = time.time()
    result = fire_release_swarm(release_id, ts_et_str)
    elapsed = time.time() - t0

    cost = result.get("usage", {}).get("billing_info", {}).get("total_cost", 0)
    print(f"{release_id} done in {elapsed:.1f}s — cost ${cost:.2f}")


# Run the whole calendar as a long-lived process
if __name__ == "__main__":
    for event in RELEASE_CALENDAR:
        schedule_release(event["release_id"], event["ts_et"])