A risk gate for a trading bot in CI

article-ci-gate.py

"""A risk gate for a trading bot, run in CI on every pull request.

The gate plays the bot on 12 markets: six ordinary ones and six with the
packaged liquidity crisis on day 50. It fails if the bot falls more than 25%
from its peak on any of them. Three versions meet it: the bot on main, a
pull request that raises its leverage from 1x to 1.5x, and a fix that
shrinks the bot's exposure when the VIX rises above 15.

In CI, set GATE_EXIT=1 and the script exits non-zero when the gate fails,
which is all a CI job needs to go red.
"""
import os
import sys

import tradefloor as tf

LIMIT = 25.0                       # the most a bot may fall from its peak, in percent
DAYS = 80
market = tf.Universe.random(20, seed=7)
MARKETS = [(seed, None) for seed in (31, 32, 33, 34, 35, 36)] + \
          [(seed, "liquidity_crisis") for seed in (41, 42, 43, 44, 45, 46)]


class Momentum:
    """Hold the five names that rose most over the last ten days, at a
    chosen gross exposure, rebalanced once a week."""

    def __init__(self, gross, vix_knee=None):
        self.gross, self.vix_knee = gross, vix_knee

    def act(self, obs):
        if not obs.is_first_step_of_day or obs.day < 10 or obs.day % 5:
            return None
        gross = self.gross
        if self.vix_knee:
            gross *= min(1.0, self.vix_knee / obs.history.macro()[-1]["vix"])
        bars = {t: obs.history.bars(t, last=10) for t in obs.tickers}
        rise = {t: b[-1]["close"] / b[0]["close"] - 1 for t, b in bars.items() if len(b) == 10}
        top = sorted(rise, key=rise.get, reverse=True)[:5]
        worth = obs.portfolio.net_worth()
        orders = {}
        for t in obs.tickers:
            want = gross * worth / 5 / obs.price(t) if t in top else 0
            change = round(want - obs.position(t))
            if change:
                orders[t] = change
        return orders or None


def gate(**bot):
    """The worst fall from a peak on each of the 12 markets."""
    falls = []
    for seed, scenario in MARKETS:
        card = tf.evaluate({"bot": Momentum(**bot)}, seed=seed, universe=market, days=DAYS,
                           history_days=10,
                           scenario=tf.Scenario.load(scenario) if scenario else None)["bot"]
        falls.append(round(card.max_drawdown_pct, 1))
    return falls


VERSIONS = {"main": dict(gross=1.0),
            "pull_request": dict(gross=1.5),
            "fix": dict(gross=1.0, vix_knee=15)}
falls = {name: gate(**bot) for name, bot in VERSIONS.items()}
failed = {name: sum(f > LIMIT for f in result) for name, result in falls.items()}
worst = {name: max(result) for name, result in falls.items()}
# The six crisis markets, for the chart.
crisis = {name: result[6:] for name, result in falls.items()}

LABELS = {"main": "main, 1x", "pull_request": "pull request, 1.5x",
          "fix": "fix, VIX brake"}
for name in VERSIONS:
    verdict = "PASS" if not failed[name] else f"FAIL on {failed[name]} of {len(MARKETS)} markets"
    print(f"{LABELS[name]:20} worst fall {worst[name]:5.1f}%  {verdict}")

if os.environ.get("GATE_EXIT"):
    sys.exit(1 if failed["pull_request"] else 0)