runner.py

import hashlib
import hmac
import json
import os
import time

import requests
from dotenv import load_dotenv
from supabase import create_client

from experiment_context import get_experiment_context, with_experiment_context

load_dotenv()

PAPER_SYMBOL = "tTESTBTC:TESTUSD"
BITFINEX_ORDER_URL_PATH = "auth/w/order/submit"
BITFINEX_ORDER_URL = f"https://api.bitfinex.com/v2/{BITFINEX_ORDER_URL_PATH}"
REQUEST_TIMEOUT_SECONDS = 20

SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
BITFINEX_API_KEY = os.getenv("BITFINEX_API_KEY")
BITFINEX_API_SECRET = os.getenv("BITFINEX_API_SECRET")
PAPER_MODE = os.getenv("PAPER_MODE", "false").lower() == "true"
SYMBOL = os.getenv("SYMBOL", PAPER_SYMBOL)


def parse_positive_int_env(name, default):
    raw_value = os.getenv(name)

    if raw_value in (None, ""):
        return default

    try:
        value = int(raw_value)
    except ValueError as exc:
        raise RuntimeError(f"{name} must be an integer.") from exc

    if value <= 0:
        raise RuntimeError(f"{name} must be greater than zero.")

    return value


def parse_positive_float_env(name, default):
    raw_value = os.getenv(name)

    if raw_value in (None, ""):
        return default

    try:
        value = float(raw_value)
    except ValueError as exc:
        raise RuntimeError(f"{name} must be a number.") from exc

    if value <= 0:
        raise RuntimeError(f"{name} must be greater than zero.")

    return value


RUNNER_POLL_SECONDS = parse_positive_int_env("RUNNER_POLL_SECONDS", 5)
DEFAULT_ORDER_AMOUNT = parse_positive_float_env("DEFAULT_ORDER_AMOUNT", 0.0001)


def fail_closed_check():
    missing = []

    for name, value in {
        "SUPABASE_URL": SUPABASE_URL,
        "SUPABASE_SERVICE_ROLE_KEY": SUPABASE_SERVICE_ROLE_KEY,
        "BITFINEX_API_KEY": BITFINEX_API_KEY,
        "BITFINEX_API_SECRET": BITFINEX_API_SECRET,
    }.items():
        if not value:
            missing.append(name)

    if missing:
        raise RuntimeError(f"Missing required env values: {', '.join(missing)}")

    if not PAPER_MODE:
        raise RuntimeError("PAPER_MODE must be true. Refusing to run.")

    if SYMBOL != PAPER_SYMBOL:
        raise RuntimeError(f"SYMBOL must be {PAPER_SYMBOL} for paper trading.")


def audit(supabase, event_type, severity="info", signal_id=None, order_id=None, payload=None):
    supabase.table("agent_audit_log").insert({
        "event_type": event_type,
        "severity": severity,
        "signal_id": signal_id,
        "order_id": order_id,
        "payload": with_experiment_context(payload),
    }).execute()


def fetch_next_signal(supabase):
    result = (
        supabase.table("agent_signals")
        .select("*")
        .eq("status", "queued")
        .order("created_at")
        .limit(1)
        .execute()
    )

    if not result.data:
        return None

    return result.data[0]


def mark_signal(supabase, signal_id, status):
    supabase.table("agent_signals").update({
        "status": status,
    }).eq("id", signal_id).execute()


def create_paper_order_record(supabase, signal):
    result = supabase.table("paper_orders").insert({
        "signal_id": signal["id"],
        "exchange": "bitfinex",
        "symbol": signal["symbol"],
        "side": signal["side"],
        "order_type": "market",
        "amount": DEFAULT_ORDER_AMOUNT,
        "status": "created",
        "request_payload": {
            "paper_mode": True,
            "source": "agent_trader_mcp_paper",
            "symbol": SYMBOL,
            "experiment": get_experiment_context(),
        },
    }).execute()

    return result.data[0]


def update_order_result(supabase, order_id, response):
    supabase.table("paper_orders").update({
        "status": "accepted",
        "exchange_order_id": response.get("exchange_order_id"),
        "response_payload": response,
    }).eq("id", order_id).execute()


def mark_order_failed(supabase, order_id, response):
    supabase.table("paper_orders").update({
        "status": "failed",
        "response_payload": response,
    }).eq("id", order_id).execute()


def submit_bitfinex_paper_order(signal, amount):
    signed_amount = amount if signal["side"] == "buy" else -amount

    body = {
        "type": "EXCHANGE MARKET",
        "symbol": signal["symbol"],
        "amount": str(signed_amount),
        "price": "0",
        "meta": {
            "source": "agent_trader_mcp_paper",
            "paper_only": True,
            "signal_id": signal["id"],
            "experiment": get_experiment_context(),
        },
    }

    body_json = json.dumps(body)
    nonce = str(int(time.time() * 1_000_000))
    signature_payload = f"/api/v2/{BITFINEX_ORDER_URL_PATH}{nonce}{body_json}"
    signature = hmac.new(
        BITFINEX_API_SECRET.encode("utf-8"),
        signature_payload.encode("utf-8"),
        hashlib.sha384,
    ).hexdigest()

    headers = {
        "bfx-nonce": nonce,
        "bfx-apikey": BITFINEX_API_KEY,
        "bfx-signature": signature,
        "content-type": "application/json",
    }

    response = requests.post(
        BITFINEX_ORDER_URL,
        headers=headers,
        data=body_json,
        timeout=REQUEST_TIMEOUT_SECONDS,
    )

    return {
        "ok": response.ok,
        "http_status": response.status_code,
        "response_text": response.text,
        "request": body,
        "symbol": signal["symbol"],
        "side": signal["side"],
        "amount": str(signed_amount),
    }


def main():
    fail_closed_check()

    supabase = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)

    print("Agent Trader MCP Paper Runner started.")
    print(f"Mode: PAPER ONLY | symbol={SYMBOL} | poll_seconds={RUNNER_POLL_SECONDS}")

    audit(
        supabase,
        "runner_started",
        payload={
            "paper_mode": PAPER_MODE,
            "symbol": SYMBOL,
            "default_order_amount": DEFAULT_ORDER_AMOUNT,
            "runner_poll_seconds": RUNNER_POLL_SECONDS,
        },
    )

    while True:
        signal_id = None
        order_id = None

        try:
            signal = fetch_next_signal(supabase)

            if not signal:
                time.sleep(RUNNER_POLL_SECONDS)
                continue

            signal_id = signal["id"]
            signal_symbol = signal.get("symbol")

            if signal_symbol != SYMBOL:
                mark_signal(supabase, signal_id, "failed")
                audit(
                    supabase,
                    "signal_rejected_disallowed_symbol",
                    severity="error",
                    signal_id=signal_id,
                    payload={
                        "expected_symbol": SYMBOL,
                        "received_symbol": signal_symbol,
                    },
                )
                print(f"Rejected signal {signal_id}: symbol {signal_symbol} is not allowed.")
                time.sleep(RUNNER_POLL_SECONDS)
                continue

            print(f"Processing signal: {signal_id} ({signal_symbol})")

            mark_signal(supabase, signal_id, "processing")
            audit(supabase, "signal_processing_started", signal_id=signal_id, payload=signal)

            order = create_paper_order_record(supabase, signal)
            order_id = order["id"]

            response = submit_bitfinex_paper_order(signal, DEFAULT_ORDER_AMOUNT)

            if response.get("ok"):
                update_order_result(supabase, order_id, response)
                mark_signal(supabase, signal_id, "executed")
                audit(
                    supabase,
                    "paper_order_executed",
                    signal_id=signal_id,
                    order_id=order_id,
                    payload=response,
                )
                print(f"Signal executed: {signal_id}")
            else:
                mark_order_failed(supabase, order_id, response)
                mark_signal(supabase, signal_id, "failed")
                audit(
                    supabase,
                    "paper_order_failed",
                    severity="error",
                    signal_id=signal_id,
                    order_id=order_id,
                    payload=response,
                )
                print(f"Signal failed: {signal_id} | http_status={response.get('http_status')}")

        except Exception as exc:
            print(f"ERROR: {exc}")

            if order_id is not None:
                try:
                    mark_order_failed(
                        supabase,
                        order_id,
                        {"ok": False, "error": str(exc)},
                    )
                except Exception as order_exc:
                    print(f"ERROR writing failed order state: {order_exc}")

            if signal_id is not None:
                try:
                    mark_signal(supabase, signal_id, "failed")
                except Exception as signal_exc:
                    print(f"ERROR writing failed signal state: {signal_exc}")

            try:
                audit(
                    supabase,
                    "runner_error",
                    severity="error",
                    signal_id=signal_id,
                    order_id=order_id,
                    payload={"error": str(exc)},
                )
            except Exception as audit_exc:
                print(f"ERROR writing runner_error audit event: {audit_exc}")

            time.sleep(RUNNER_POLL_SECONDS)


if __name__ == "__main__":
    main()