Plugin

Webhook summary

Posts a run's headline figures to a webhook of your own, as generic JSON, a Discord message or a Slack message. HTTPS only, summary only by default.

ExporterNetworkDesktop onlyv1.0.0

Get early access Plugin API documentation

About

After a run, the exporter sends net P&L, return, maximum drawdown, the number of trades, the win rate and the profit factor, with the instrument, timeframe and period, to a webhook you set up: your own service or an automation tool (generic JSON), a Discord channel or a Slack channel. The trades go along only when you switch them on.

Nothing leaves your machine by accident. You enter the webhook's host as a setting of its own, so you see where the summary goes, and the URL has to be on exactly that host and use HTTPS. While Send is off, which is the default, an export sends nothing: it shows the payload, the host it would go to and whether the settings are complete. Switch Send on once that looks right. The webhook URL is a secret, so neither the preview nor the plugin's log shows more of it than the host.

The generic JSON says "format": "orionquant-webhook-summary", "formatVersion": 1 and "hypothetical": true, with the blocks run and summary, and trades (at most 500, the rest counted in tradesOmitted) when switched on. Discord and Slack get a short text message; trades that do not fit their length limit are counted at the end.

Where to get a webhook URL: Discord, in a channel's settings under Integrations, Webhooks (https://docs.discord.com/developers/resources/webhook); Slack, with an app that has incoming webhooks (https://docs.slack.dev/messaging/sending-messages-using-incoming-webhooks). Their terms apply to what you post: Discord's developer terms and Slack's API terms. Slack allows about one message per second per webhook; each export sends one.

Configuration

Field	Type	Default	Description
`format`Payload format	select	`"generic"`	generic: a JSON object for your own service. discord: a Discord webhook message (mentions switched off). slack: a Slack incoming webhook message.Options: generic, discord, slack
`host`Webhook host (where it goes)	string	`""`	The one host this plugin may send to, like discord.com or hooks.slack.com. The webhook URL must be on it.
`includeTrades`Include the trades	boolean	`false`	Send the trades too (time, side, symbol, prices, lots, net, exit reason). Off sends the summary only.
`label`Message title	string	`"OrionQuant backtest"`	The title of the message, so you can tell runs apart in a channel.
`send`Send (off shows a preview)	boolean	`false`	Off: an export only shows what it would send and where. On: an export posts it.
`webhookUrl`Webhook URL (secret)	string	`""`	The webhook's full https:// URL from Discord, Slack or your service. Treat it like a password: whoever has it can post to your channel.

Use it in a strategy

Not called from a strategy: after a run, choose it in the result's export menu. Exports are part of the plans that include CSV export.

What it provides

`POST <webhook URL>`Payload	One HTTPS POST with a JSON body in the chosen format, when Send is on. The export answers with the host and the HTTP status.
`preview`Payload	While Send is off: the payload, the host it would go to, and whether the settings are ready, as text. Nothing is sent.

Limits

It sends to the host of the URL you enter, which can be any HTTPS host; the host setting has to name the same host, so you see it before anything is sent.
The webhook URL is kept in the plugin's settings on this machine, in plain text like every plugin setting.
One attempt per export, no retry and no redirect: a webhook that is down or moved means the export fails and says why.
The figures are those of a hypothetical backtest; every message says so.

Preview

json{
  "format": "orionquant-webhook-summary",
  "formatVersion": 1,
  "hypothetical": true,
  "label": "OrionQuant backtest",
  "run": {
    "currency": "USD",
    "endDate": "2024-06-28",
    "instrument": "EURUSD",
    "seed": 42,
    "startDate": "2024-01-02",
    "timeframe": "H1"
  },
  "source": "OrionQuant",
  "summary": {
    "maxDrawdown": 410.25,
    "maxDrawdownPct": 3.81,
    "netPnl": 1250.4,
    "profitFactor": 1.62,
    "returnPct": 12.5,
    "trades": 48,
    "winRate": 54.2
  }
}

A made-up sample that shows the shape of the output. Not vendor data and not a result.

The code

main.py, the file the app installs.

python"""Webhook summary: post a run's headline figures to a webhook of your own.

Three payload shapes: generic JSON for your own service or an automation
tool, a Discord message, or a Slack message. The summary is net P&L, return,
maximum drawdown, the number of trades, the win rate and the profit factor,
with the run's instrument, timeframe and period. The trades go along only
when you switch them on.

Sending is a deliberate act in two steps. The webhook's host is a setting of
its own, next to the URL, so you see where the summary goes before anything
leaves your machine, and the URL must be on exactly that host and use
HTTPS. And while "Send" is off (the default), an export sends nothing: it
hands back what it would send, to whom, and whether the settings are ready.
Switch "Send" on once that looks right.

The webhook URL is a secret (whoever has it can post to your channel), so it
is never written to the plugin's log or to the preview: both show the host
only.

The POST is the kit's net.post_json: HTTPS only, the URL's own host as the
one allowed host (plugin.json declares the URL setting under
network.userHosts), a timeout, no redirects and no retries. Any answer that
is not a 2xx is a failed export that names the host and the status.
"""

import json
import math

from orionquant_engine.plugins.kit import net

_API_REF = {"api": None}

FORMATS = ("generic", "discord", "slack")

#: The most trades the generic payload carries; the rest are counted, not sent.
MAX_TRADES = 500

#: Characters a chat message may use. Discord refuses content over 2000;
#: Slack takes far more but asks for short messages, and a summary is one.
_TEXT_LIMITS = {"discord": 2000, "slack": 3000}

#: Seconds to wait for the webhook to answer.
TIMEOUT = 15.0


def _setting(key, default):
    api = _API_REF["api"]
    return api.get_config(key, default) if api else default


def _num(value):
    """A finite float, or None (JSON has no Infinity; a profit factor can be)."""
    if value is None or isinstance(value, bool):
        return None
    try:
        f = float(value)
    except (TypeError, ValueError):
        return None
    return f if math.isfinite(f) else None


def _round(value, places):
    v = _num(value)
    return None if v is None else round(v, places)


def summary(result):
    """The headline figures of a run, as plain numbers."""
    stats = result.get("stats") or {}
    return {
        "netPnl": _round(stats.get("netPnl"), 2),
        "returnPct": _round(stats.get("returnPct"), 2),
        "maxDrawdown": _round(stats.get("maxDrawdown"), 2),
        "maxDrawdownPct": _round(stats.get("maxDrawdownPct"), 2),
        "trades": int(stats.get("numTrades") or len(result.get("trades") or [])),
        "winRate": _round(stats.get("winRate"), 1),
        "profitFactor": _round(stats.get("profitFactor"), 2),
    }


def run_facts(result):
    config = result.get("config") or {}
    return {
        "instrument": config.get("instrument", ""),
        "timeframe": config.get("timeframe", ""),
        "startDate": config.get("startDate", ""),
        "endDate": config.get("endDate", ""),
        "seed": result.get("seed"),
        "currency": result.get("resultCurrency", ""),
    }


def _trade(t):
    """The fields of one trade a webhook gets: when, which way, how it ended."""
    return {
        "id": t.get("id"),
        "symbol": t.get("symbol"),
        "direction": t.get("direction"),
        "entryTs": t.get("entryTs"),
        "exitTs": t.get("exitTs"),
        "entryPrice": _num(t.get("entryPrice")),
        "exitPrice": _num(t.get("exitPrice")),
        "lots": _num(t.get("quantityLots")),
        "netPnl": _round(t.get("netPnl"), 2),
        "exitReason": t.get("exitReason"),
    }


def _money(value, currency):
    return "n/a" if value is None else "{:,.2f} {}".format(value, currency).strip()


def _pct(value, places=2):
    return "n/a" if value is None else "{:.{}f}%".format(value, places)


def _chat_lines(result, label, include_trades, bold, esc):
    """(head lines, trade lines) of a chat message. `esc` escapes text for the
    chat's markup, `bold` then wraps the title in it."""
    run, s = run_facts(result), summary(result)
    cur = run["currency"]
    head = [
        "{} {}, {} to {}".format(run["instrument"], run["timeframe"], run["startDate"], run["endDate"]),
        "Net: {} ({})".format(_money(s["netPnl"], cur), _pct(s["returnPct"])),
        "Max drawdown: {} ({})".format(_money(s["maxDrawdown"], cur), _pct(s["maxDrawdownPct"])),
        "Trades: {}, win rate {}, profit factor {}".format(
            s["trades"], _pct(s["winRate"], 1), "n/a" if s["profitFactor"] is None else s["profitFactor"]),
    ]
    head = [bold(esc(label)) + " (hypothetical backtest)"] + [esc(x) for x in head]
    trades = []
    if include_trades:
        for t in result.get("trades") or []:
            trades.append("#{} {} {} {} to {}: {} ({})".format(
                t.get("id"), t.get("direction"), t.get("symbol"), str(t.get("entryTs", ""))[:16].replace("T", " "),
                str(t.get("exitTs", ""))[:16].replace("T", " "), _money(_round(t.get("netPnl"), 2), cur),
                str(t.get("exitReason", "")).lower()))
    return head, [esc(x) for x in trades]


def _fit(head, trades, limit):
    """The message text: the head, then as many trade lines as fit in `limit`."""
    text = "\n".join(head)
    for i, line in enumerate(trades):
        rest = len(trades) - i - 1
        tail = "\n... and {} more trades".format(rest) if rest else ""
        if len(text) + 1 + len(line) + len(tail) > limit:
            return text + "\n... and {} more trades".format(len(trades) - i)
        text += "\n" + line
    return text


def _slack_escape(text):
    """Slack reads &, < and > as markup; these three are all it asks to escape."""
    return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


def build_payload(result, *, fmt="generic", label="OrionQuant backtest", include_trades=False):
    """The JSON body for one format, as a dict."""
    if fmt not in FORMATS:
        raise ValueError("format must be one of {}, not {!r}".format(", ".join(FORMATS), fmt))
    if fmt == "generic":
        payload = {
            "source": "OrionQuant",
            "format": "orionquant-webhook-summary",
            "formatVersion": 1,
            "label": label,
            "hypothetical": True,
            "run": run_facts(result),
            "summary": summary(result),
        }
        if include_trades:
            trades = list(result.get("trades") or [])
            payload["trades"] = [_trade(t) for t in trades[:MAX_TRADES]]
            payload["tradesOmitted"] = max(0, len(trades) - MAX_TRADES)
        return payload
    if fmt == "discord":
        head, trades = _chat_lines(result, label, include_trades, lambda s: "**{}**".format(s), lambda s: s)
        # No @everyone, no role or user pings, whatever the label or a symbol says.
        return {"content": _fit(head, trades, _TEXT_LIMITS["discord"]), "allowed_mentions": {"parse": []}}
    head, trades = _chat_lines(result, label, include_trades, lambda s: "*{}*".format(s), _slack_escape)
    return {"text": _fit(head, trades, _TEXT_LIMITS["slack"])}


def where(url, host):
    """(problem or None, the URL as shown: host only, never the secret path)."""
    url, host = str(url or "").strip(), str(host or "").strip().lower()
    if not host:
        return "Set the webhook host first: the one host this plugin may send to.", ""
    if not url:
        return "Set the webhook URL first.", "https://{}/...".format(host)
    try:
        net.check_url(url, [net.host_of(url)])
        if net.host_of(url) != host:
            return "The webhook URL is not on {}: the URL and the host setting must agree.".format(host), ""
    except net.NetError as exc:
        if exc.code == "host_not_allowed":
            return "The webhook URL is not on {}: the URL and the host setting must agree.".format(host), ""
        if exc.code == "not_https":
            return "The webhook URL must use HTTPS.", ""
        return str(exc).replace(url, "the webhook URL"), ""
    return None, "https://{}/...".format(host)


def _send(url, host, payload):
    """POST the payload through the kit; the HTTP status of a 2xx answer.

    Every failure is an error naming the host, never the URL: the URL is the
    secret, and an error message ends up in the app and in the log.
    """
    try:
        status, _text = net.post_json(url, payload, hosts=[net.host_of(url)], timeout=TIMEOUT)
    except net.NetError as exc:
        raise net.NetError(exc.code, str(exc).replace(url, host), status=exc.status) from None
    if not 200 <= status < 300:
        raise net.NetError("http_status", "{} answered HTTP {}".format(host, status), status=status)
    return status


def export(result):
    fmt = str(_setting("format", "generic"))
    url = str(_setting("webhookUrl", "") or "").strip()
    host = str(_setting("host", "") or "").strip().lower()
    payload = build_payload(result, fmt=fmt, label=str(_setting("label", "OrionQuant backtest")),
                            include_trades=bool(_setting("includeTrades", False)))
    problem, shown = where(url, host)
    api = _API_REF["api"]

    if not _setting("send", False):
        preview = {
            "sent": False,
            "why": "Send is off: nothing left this machine. Switch Send on to post this.",
            "to": shown,
            "host": host,
            "format": fmt,
            "ready": problem is None,
            "problem": problem,
            "payload": payload,
        }
        return json.dumps(preview, indent=2, ensure_ascii=False, allow_nan=False)

    if problem:
        raise ValueError(problem)
    status = _send(url, host, payload)
    if api:
        api.log("posted the summary to {} ({}, HTTP {})".format(host, fmt, status))
    return "Sent the {} summary to {} (HTTP {}).".format(fmt, host, status)


def register(api):
    _API_REF["api"] = api
    api.register_exporter("webhook_summary", "Webhook summary", export)