Python API Reference

Top-level exports

import balancelab

balancelab

balancelab - adversarial game economy red-team library.

ConfidenceSequenceState(n, mean, half_width, lower, upper, total_cost, decisive, precision_met, sign_settled) dataclass

Streaming anytime-valid style bounds on mean value.

Attributes:

Name	Type	Description
n	int	Number of observations.
mean	float	Sample mean of values.
half_width	float	± half-width of the sequence at this n.
lower	/ upper	mean ± half_width.
total_cost	float	Sum of observation costs.
decisive	bool	True when 0 is outside [lower, upper] (sign settled) or half_width ≤ target_precision and n ≥ min_n (precision met).
precision_met	bool	half_width ≤ target_precision (with enough samples).
sign_settled	bool	lower > 0 or upper < 0.

EvalObservation(value, cost=1.0, label='') dataclass

One paired evaluation outcome (agent A minus agent B, or signed score).

ClosedLoopError

Bases: ValueError

Raised when the economy/signal gate refuses empty or unsafe state.

GateOutcome(ok, verdict, reason, exit_code, rule_count=0, exploit_count=0, max_gain_ratio=None, human_required=False) dataclass

Result of a closed-loop economy or signal gate.

Attributes:

Name	Type	Description
ok	bool	True only when ship/run may continue.
verdict	str	PASS, FAIL, or FAIL_LOUD.
reason	str	Always non-empty.
exit_code	int	0 PASS, 1 FAIL (exploit/signal), 2 FAIL_LOUD (empty).
rule_count	int	Economy rules examined.
exploit_count	int	Exploits found.
max_gain_ratio	float | None	Largest exploit gain if any.
human_required	bool	True when design review is required.

EconomyGraph(rules=list()) dataclass

A directed exchange graph of EconomyRules.

add_rule(rule)

Add a rule to the graph.

Note: duplicate rules (same source, target, and quantities) produce the same content-addressed id and will be stored as separate entries. Callers that want idempotent insertion should check graph.rules first.

Source code in src/balancelab/economy.py

def add_rule(self, rule: EconomyRule) -> None:
    """Add a rule to the graph.

    Note: duplicate rules (same source, target, and quantities) produce the
    same content-addressed ``id`` and will be stored as separate entries.
    Callers that want idempotent insertion should check ``graph.rules`` first.
    """
    self.rules.append(rule)

neighbors(item)

Return [(target_item, rule)] for all rules from item.

Source code in src/balancelab/economy.py

def neighbors(self, item: str) -> list[tuple[str, EconomyRule]]:
    """Return [(target_item, rule)] for all rules from item."""
    return [(r.target_item, r) for r in self.rules if r.source_item == item]

items()

All item names in the graph.

Source code in src/balancelab/economy.py

def items(self) -> set[str]:
    """All item names in the graph."""
    result: set[str] = set()
    for r in self.rules:
        result.add(r.source_item)
        result.add(r.target_item)
    return result

to_dict()

Serialize to dict.

Source code in src/balancelab/economy.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to dict."""
    return {
        "rules": [r.to_dict() for r in self.rules],
    }

EconomyRule(source_item, target_item, source_qty, target_qty, rule_id='', tags=list()) dataclass

A single exchange rule in the economy.

exchange_rate()

Return target_qty / source_qty.

Source code in src/balancelab/economy.py

def exchange_rate(self) -> float:
    """Return target_qty / source_qty."""
    return self.target_qty / self.source_qty if self.source_qty > 0 else 0.0

to_dict()

Serialize to dict.

Source code in src/balancelab/economy.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to dict."""
    return {
        "id": self.id,
        "source_item": self.source_item,
        "target_item": self.target_item,
        "source_qty": self.source_qty,
        "target_qty": self.target_qty,
        "rule_id": self.rule_id,
        "tags": self.tags,
    }

from_dict(d) classmethod

Deserialize from dict.

Source code in src/balancelab/economy.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> EconomyRule:
    """Deserialize from dict."""
    r = cls(
        source_item=d["source_item"],
        target_item=d["target_item"],
        source_qty=d["source_qty"],
        target_qty=d["target_qty"],
        rule_id=d.get("rule_id", ""),
        tags=d.get("tags", []),
    )
    return r

ExploitFinder

Find arbitrage exploits using Bellman-Ford on log-weight graph.

find_exploits(graph)

Convert exchange rates to log-weights: weight = -log(rate). Negative cycles in the log-weight graph = positive gain cycles. Use Bellman-Ford to detect negative cycles. Return ExploitReport with all found cycles.

Source code in src/balancelab/economy.py

def find_exploits(self, graph: EconomyGraph) -> ExploitReport:
    """
    Convert exchange rates to log-weights: weight = -log(rate).
    Negative cycles in the log-weight graph = positive gain cycles.
    Use Bellman-Ford to detect negative cycles.
    Return ExploitReport with all found cycles.
    """
    items = list(graph.items())
    n = len(items)
    if n == 0:
        return ExploitReport(
            graph_item_count=0,
            graph_rule_count=len(graph.rules),
            exploits=[],
            total_found=0,
        )

    item_idx = {item: i for i, item in enumerate(items)}

    # Build log-weight adjacency: weight = -log(exchange_rate)
    # Negative cycle = positive gain cycle
    edges = []
    for rule in graph.rules:
        rate = rule.exchange_rate()
        if rate > 0:
            weight = -math.log(rate)
            edges.append((item_idx[rule.source_item], item_idx[rule.target_item], weight, rule))

    exploits = []
    # Try Bellman-Ford from each source to find negative cycles
    for start in range(n):
        dist = [float("inf")] * n
        pred: list[int | None] = [None] * n
        pred_rule: list[EconomyRule | None] = [None] * n
        dist[start] = 0.0

        for _ in range(n - 1):
            for u, v, w, rule in edges:
                if dist[u] != float("inf") and dist[u] + w < dist[v]:
                    dist[v] = dist[u] + w
                    pred[v] = u
                    pred_rule[v] = rule

        # Check for negative cycles
        for u, v, w, _rule in edges:
            if dist[u] != float("inf") and dist[u] + w < dist[v]:
                # Found negative cycle - trace it back
                cycle_nodes: list[str] = []
                cycle_rules: list[str] = []
                visited: set[int] = set()
                curr = v
                # advance enough steps to ensure we're in the cycle
                for _ in range(n):
                    curr = pred[curr]  # type: ignore[assignment]
                    if curr is None:
                        break
                # now trace the cycle
                if curr is None:
                    break
                cycle_start = curr
                curr = cycle_start
                while True:
                    visited.add(curr)
                    cycle_nodes.append(items[curr])
                    if pred_rule[curr] is not None:
                        cycle_rules.append(pred_rule[curr].id)  # type: ignore[union-attr]
                    curr = pred[curr]  # type: ignore[assignment]
                    if curr is None or curr == cycle_start:
                        break

                # close the cycle
                cycle_nodes.append(items[cycle_start])

                # Reverse: path was built following predecessor links (backwards),
                # so reverse to get actual rule-flow direction (e.g. gold→silver→gems→gold).
                cycle_nodes = cycle_nodes[::-1]
                cycle_rules = cycle_rules[::-1]

                # Compute gain ratio - multiply exchange rates around the cycle
                gain = 1.0
                for node_idx in visited:
                    if pred_rule[node_idx] is not None:
                        gain *= pred_rule[node_idx].exchange_rate()  # type: ignore[union-attr]

                if gain > 1.0:
                    exploit = ExploitPath(
                        path=cycle_nodes,
                        rules_used=cycle_rules,
                        gain_ratio=gain,
                    )
                    exploits.append(exploit)
                break

    # Deduplicate exploits by path set
    seen: set[frozenset[str]] = set()
    unique: list[ExploitPath] = []
    for e in exploits:
        key = frozenset(e.path)
        if key not in seen:
            seen.add(key)
            unique.append(e)

    return ExploitReport(
        graph_item_count=len(items),
        graph_rule_count=len(graph.rules),
        exploits=unique,
        total_found=len(unique),
    )

ExploitPath(path, rules_used, gain_ratio) dataclass

A circular path that yields net gain.

to_dict()

Serialize to dict.

Source code in src/balancelab/economy.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to dict."""
    return {
        "id": self.id,
        "path": self.path,
        "rules_used": self.rules_used,
        "gain_ratio": self.gain_ratio,
    }

ExploitReport(graph_item_count, graph_rule_count, exploits, total_found, timestamp=(lambda: time.time())()) dataclass

Report of all exploits found in an economy.

to_dict()

Serialize to dict.

Source code in src/balancelab/economy.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to dict."""
    return {
        "id": self.id,
        "graph_item_count": self.graph_item_count,
        "graph_rule_count": self.graph_rule_count,
        "exploits": [e.to_dict() for e in self.exploits],
        "total_found": self.total_found,
        "timestamp": self.timestamp,
    }

assert_eval_stopping_ok(observations, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_eval_stopping is ok.

Source code in src/balancelab/av_aivat.py

def assert_eval_stopping_ok(
    observations: Sequence[EvalObservation | dict[str, Any] | float | int] | None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_eval_stopping` is ok."""
    outcome = gate_eval_stopping(observations, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

gate_eval_stopping(observations, *, decision, target_precision=DEFAULT_TARGET_PRECISION, z=DEFAULT_Z, min_n=2, sample_sd=None, max_total_cost=None, require_observations=True)

Refuse wasteful continue or premature stop (AV-AIVAT class).

Rules:

• No observations when required → FAIL_LOUD
• decision=continue while sequence is decisive → FAIL (keep paying after result settled - paper failure mode)
• decision=stop while sequence is not decisive → FAIL (stop before agents can be told apart / precision unmet)
• max_total_cost exceeded and still continuing → FAIL
• continue while not decisive (and under budget) → PASS
• stop while decisive → PASS

Parameters:

Name	Type	Description	Default
observations	Sequence[EvalObservation | dict[str, Any] | float | int] | None	Stream of paired eval values so far.	required
decision	Decision	Proposed next action for the evaluation loop.	required
target_precision	float	Half-width goal (paper ±1 BB class).	DEFAULT_TARGET_PRECISION
z	float	Critical value for the plug-in CS half-width.	DEFAULT_Z
min_n	int	Minimum samples before precision_met can fire.	2
sample_sd	float | None	Optional known SD (else sample).	None
max_total_cost	float | None	Optional hard budget ceiling on continue.	None
require_observations	bool	Empty stream → FAIL_LOUD when True.	True

Source code in src/balancelab/av_aivat.py

def gate_eval_stopping(
    observations: Sequence[EvalObservation | dict[str, Any] | float | int] | None,
    *,
    decision: Decision,
    target_precision: float = DEFAULT_TARGET_PRECISION,
    z: float = DEFAULT_Z,
    min_n: int = 2,
    sample_sd: float | None = None,
    max_total_cost: float | None = None,
    require_observations: bool = True,
) -> GateOutcome:
    """Refuse wasteful continue or premature stop (AV-AIVAT class).

    Rules:

    * No observations when required → **FAIL_LOUD**
    * ``decision=continue`` while sequence is **decisive** → **FAIL**
      (keep paying after result settled - paper failure mode)
    * ``decision=stop`` while sequence is **not decisive** → **FAIL**
      (stop before agents can be told apart / precision unmet)
    * ``max_total_cost`` exceeded and still continuing → **FAIL**
    * continue while not decisive (and under budget) → **PASS**
    * stop while decisive → **PASS**

    Args:
        observations: Stream of paired eval values so far.
        decision: Proposed next action for the evaluation loop.
        target_precision: Half-width goal (paper ±1 BB class).
        z: Critical value for the plug-in CS half-width.
        min_n: Minimum samples before precision_met can fire.
        sample_sd: Optional known SD (else sample).
        max_total_cost: Optional hard budget ceiling on continue.
        require_observations: Empty stream → FAIL_LOUD when True.
    """
    dec = (decision or "").strip().lower()
    if dec not in {"continue", "stop"}:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=(f"AV-AIVAT: unknown decision={decision!r} (use continue|stop)"),
            exit_code=2,
            human_required=True,
        )

    if not observations:
        if require_observations:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=(
                    "AV-AIVAT: no evaluation observations - cannot decide "
                    "continue/stop without a streaming score inventory "
                    "(arXiv 2608.06362)"
                ),
                exit_code=2,
                human_required=True,
            )
        if dec == "stop":
            return GateOutcome(
                ok=True,
                verdict="PASS",
                reason="AV-AIVAT: no observations required; stop allowed",
                exit_code=0,
                human_required=False,
            )
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason="AV-AIVAT: continue with empty stream refused",
            exit_code=2,
            human_required=True,
        )

    try:
        state = summarize_confidence_sequence(
            observations,
            target_precision=target_precision,
            z=z,
            min_n=min_n,
            sample_sd=sample_sd,
        )
    except (TypeError, ValueError) as exc:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=f"AV-AIVAT: invalid observations: {exc}",
            exit_code=2,
            human_required=True,
        )

    if max_total_cost is not None and state.total_cost > max_total_cost and dec == "continue":
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"AV-AIVAT: total_cost={state.total_cost:.3f} exceeds "
                f"max_total_cost={max_total_cost:.3f} while decision=continue - "
                "budget exhausted (token/game spend runaway)"
            ),
            exit_code=1,
            human_required=True,
            rule_count=state.n,
            max_gain_ratio=state.mean,
        )

    if dec == "continue" and state.decisive:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"AV-AIVAT: decision=continue but sequence already decisive "
                f"(n={state.n} mean={state.mean:.4f} "
                f"CI=[{state.lower:.4f},{state.upper:.4f}] "
                f"sign_settled={state.sign_settled} "
                f"precision_met={state.precision_met}) - refuse keep-paying "
                "after result settled (arXiv 2608.06362 fixed-budget waste)"
            ),
            exit_code=1,
            human_required=False,
            rule_count=state.n,
            max_gain_ratio=state.mean,
        )

    if dec == "stop" and not state.decisive:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"AV-AIVAT: decision=stop but sequence not decisive "
                f"(n={state.n} mean={state.mean:.4f} "
                f"half_width={state.half_width:.4f} "
                f"target={target_precision}) - refuse premature stop before "
                "agents can be told apart / precision unmet"
            ),
            exit_code=1,
            human_required=True,
            rule_count=state.n,
            max_gain_ratio=state.mean,
        )

    if dec == "stop":
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=(
                f"AV-AIVAT ok: stop with decisive sequence n={state.n} "
                f"mean={state.mean:.4f} half_width={state.half_width:.4f} "
                f"cost={state.total_cost:.3f}"
            ),
            exit_code=0,
            human_required=False,
            rule_count=state.n,
            max_gain_ratio=state.mean,
        )

    # continue, not decisive
    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"AV-AIVAT ok: continue n={state.n} mean={state.mean:.4f} "
            f"half_width={state.half_width:.4f} target={target_precision} "
            f"cost={state.total_cost:.3f}"
        ),
        exit_code=0,
        human_required=False,
        rule_count=state.n,
        max_gain_ratio=state.mean,
    )

summarize_confidence_sequence(observations, *, target_precision=DEFAULT_TARGET_PRECISION, z=DEFAULT_Z, min_n=2, sample_sd=None)

Compute a streaming Gaussian-style confidence sequence snapshot.

Uses half_width = z * s / sqrt(n) with sample SD (or provided sample_sd). This is a gate-facing plug-in CS, not a claim of exact AV-AIVAT AIVAT corrections - those reduce variance upstream; the gate consumes the resulting stream of values.

Source code in src/balancelab/av_aivat.py

def summarize_confidence_sequence(
    observations: Sequence[EvalObservation | dict[str, Any] | float | int],
    *,
    target_precision: float = DEFAULT_TARGET_PRECISION,
    z: float = DEFAULT_Z,
    min_n: int = 2,
    sample_sd: float | None = None,
) -> ConfidenceSequenceState:
    """Compute a streaming Gaussian-style confidence sequence snapshot.

    Uses ``half_width = z * s / sqrt(n)`` with sample SD (or provided
    ``sample_sd``). This is a **gate-facing** plug-in CS, not a claim of
    exact AV-AIVAT AIVAT corrections - those reduce variance upstream; the
    gate consumes the resulting stream of values.
    """
    if target_precision < 0:
        raise ValueError("target_precision must be >= 0")
    if z <= 0:
        raise ValueError("z must be > 0")

    obs = [_as_obs(x) for x in observations]
    n = len(obs)
    if n == 0:
        return ConfidenceSequenceState(
            n=0,
            mean=0.0,
            half_width=float("inf"),
            lower=float("-inf"),
            upper=float("inf"),
            total_cost=0.0,
            decisive=False,
            precision_met=False,
            sign_settled=False,
        )

    values = [o.value for o in obs]
    mean = sum(values) / n
    total_cost = sum(o.cost for o in obs)

    if n == 1:
        s = abs(sample_sd) if sample_sd is not None else abs(values[0]) or 1.0
    elif sample_sd is not None:
        s = abs(float(sample_sd))
    else:
        var = sum((v - mean) ** 2 for v in values) / max(n - 1, 1)
        s = math.sqrt(var) if var > 0 else 0.0
        # floor SD so empty variance still widens early stops
        if s == 0.0:
            s = 1e-9

    half = z * s / math.sqrt(n)
    lower = mean - half
    upper = mean + half
    sign_settled = lower > 0.0 or upper < 0.0
    precision_met = n >= min_n and half <= target_precision
    decisive = sign_settled or precision_met

    return ConfidenceSequenceState(
        n=n,
        mean=mean,
        half_width=half,
        lower=lower,
        upper=upper,
        total_cost=total_cost,
        decisive=decisive,
        precision_met=precision_met,
        sign_settled=sign_settled,
    )

gate_binary_signal(direction, yes_price, no_price=None, *, down_uses_no_token=True)

Gate binary market signal mapping (SIGNAL-INVERT farm case).

Farm failure: BTC DOWN used DOWN mid as YES price on the DOWN contract - inverted directional signal.

For a DOWN signal on a market where YES = "price goes down": * correct: use YES price of the DOWN contract * invert trap: treating "DOWN mid" as if it were the UP/YES without checking token polarity

This gate checks that for direction=down, yes_price is the probability mass for the DOWN outcome (typically lower when market is bullish). When both yes and no are provided, they must sum ~1 and direction must pick the cheaper/correct side consistently.

Parameters:

Name	Type	Description	Default
direction	str	up or down (case-insensitive).	required
yes_price	float	Price used as YES for the trade decision.	required
no_price	float | None	Optional NO price for consistency check.	None
down_uses_no_token	bool	If True, direction=down must not use a yes_price that is mislabeled - when no_price given and direction=down, the traded token price should be yes_price only if YES means down; we require yes_price + no_price ≈ 1 and 0 < prices < 1.	True

Source code in src/balancelab/closed_loop.py

def gate_binary_signal(
    direction: str,
    yes_price: float,
    no_price: float | None = None,
    *,
    down_uses_no_token: bool = True,
) -> GateOutcome:
    """Gate binary market signal mapping (SIGNAL-INVERT farm case).

    Farm failure: BTC DOWN used DOWN mid as YES price on the DOWN contract -
    inverted directional signal.

    For a DOWN signal on a market where YES = "price goes down":
      * correct: use YES price of the DOWN contract
      * invert trap: treating "DOWN mid" as if it were the UP/YES without
        checking token polarity

    This gate checks that for ``direction=down``, ``yes_price`` is the
    probability mass for the DOWN outcome (typically lower when market is
    bullish). When both yes and no are provided, they must sum ~1 and
    direction must pick the cheaper/correct side consistently.

    Args:
        direction: ``up`` or ``down`` (case-insensitive).
        yes_price: Price used as YES for the trade decision.
        no_price: Optional NO price for consistency check.
        down_uses_no_token: If True, direction=down must not use a yes_price
            that is mislabeled - when no_price given and direction=down,
            the traded token price should be yes_price only if YES means down;
            we require yes_price + no_price ≈ 1 and 0 < prices < 1.
    """
    d = (direction or "").strip().lower()
    if d not in {"up", "down", "yes", "no"}:
        return _fail_loud(f"SIGNAL-INVERT: unknown direction {direction!r}")

    if not (0.0 < yes_price < 1.0):
        return _fail(
            f"SIGNAL-INVERT: yes_price={yes_price} out of (0,1) - invalid quote",
            rule_count=1,
        )

    if no_price is not None:
        if not (0.0 < no_price < 1.0):
            return _fail(
                f"SIGNAL-INVERT: no_price={no_price} out of (0,1)",
                rule_count=2,
            )
        s = yes_price + no_price
        if abs(s - 1.0) > 0.05:
            return _fail(
                f"SIGNAL-INVERT: yes+no={s:.4f} not ~1.0 - inverted/wrong legs "
                f"(farm: DOWN mid as YES)",
                rule_count=2,
            )
        # Classic invert: using the expensive wrong leg for direction
        if d == "down" and down_uses_no_token and yes_price > no_price + 0.02:
            # When YES is the UP token, DOWN should trade NO (price = no_price).
            # If agent passes yes_price while intending down, flag if yes > no
            # (buying the expensive wrong side for a down view).
            return _fail(
                f"SIGNAL-INVERT: direction=down but yes_price={yes_price:.4f} > "
                f"no_price={no_price:.4f} - likely using UP/YES as DOWN signal "
                f"(farm: BTC DOWN mid as YES)",
                rule_count=2,
            )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=f"binary signal ok direction={d} yes={yes_price}",
        exit_code=0,
        rule_count=1 if no_price is None else 2,
        human_required=False,
    )

gate_economy(graph, *, finder=None, max_allowed_gain=1.0, min_rules=1)

Scan economy for exploits - load-bearing ship/no-ship gate.

• Empty rules → FAIL_LOUD
• Any exploit with gain_ratio > max_allowed_gain → FAIL (no-ship)
• Clean graph → PASS

Parameters:

Name	Type	Description	Default
graph	EconomyGraph	Economy exchange graph.	required
finder	ExploitFinder | None	Optional ExploitFinder (default new instance).	None
max_allowed_gain	float	Gains strictly above this fail (default 1.0 = any profit).	1.0
min_rules	int	Minimum rules required.	1

Source code in src/balancelab/closed_loop.py

def gate_economy(
    graph: EconomyGraph,
    *,
    finder: ExploitFinder | None = None,
    max_allowed_gain: float = 1.0,
    min_rules: int = 1,
) -> GateOutcome:
    """Scan economy for exploits - load-bearing ship/no-ship gate.

    * Empty rules → FAIL_LOUD
    * Any exploit with gain_ratio > max_allowed_gain → FAIL (no-ship)
    * Clean graph → PASS

    Args:
        graph: Economy exchange graph.
        finder: Optional ExploitFinder (default new instance).
        max_allowed_gain: Gains strictly above this fail (default 1.0 = any profit).
        min_rules: Minimum rules required.
    """
    n = len(graph.rules)
    if n < min_rules:
        return _fail_loud(
            f"empty economy - {n} rules (<{min_rules}); cannot gate a phantom graph",
            rule_count=n,
            exploit_count=0,
        )

    report = (finder or ExploitFinder()).find_exploits(graph)
    return gate_exploit_report(report, max_allowed_gain=max_allowed_gain)

gate_kill_switch(trade_pnls, loss_limit, *, paper_mode=False, max_trip_rate=0.5)

Gate kill-switch configuration (KILL-SWITCH farm case).

Farm failure: loss-limit tripped on every paper fill (worst-case spread model) - strategy could not be evaluated.

Parameters:

Name	Type	Description	Default
trade_pnls	Sequence[float]	Per-trade PnL series (negative = loss).	required
loss_limit	float	Absolute loss that trips the switch (positive number).	required
paper_mode	bool	If True, apply stricter trip-rate limits.	False
max_trip_rate	float	Fail if fraction of trades that would trip exceeds this.	0.5

Source code in src/balancelab/closed_loop.py

def gate_kill_switch(
    trade_pnls: Sequence[float],
    loss_limit: float,
    *,
    paper_mode: bool = False,
    max_trip_rate: float = 0.5,
) -> GateOutcome:
    """Gate kill-switch configuration (KILL-SWITCH farm case).

    Farm failure: loss-limit tripped on every paper fill (worst-case spread
    model) - strategy could not be evaluated.

    Args:
        trade_pnls: Per-trade PnL series (negative = loss).
        loss_limit: Absolute loss that trips the switch (positive number).
        paper_mode: If True, apply stricter trip-rate limits.
        max_trip_rate: Fail if fraction of trades that would trip exceeds this.
    """
    if loss_limit <= 0:
        return _fail_loud(
            f"KILL-SWITCH: invalid loss_limit={loss_limit} (must be > 0)",
        )

    if not trade_pnls:
        return _fail_loud("KILL-SWITCH: empty trade PnL series - nothing to gate")

    # Trip if cumulative or single trade exceeds limit
    trips = 0
    cum = 0.0
    for pnl in trade_pnls:
        cum += pnl
        if pnl <= -loss_limit or cum <= -loss_limit:
            trips += 1
            # reset cum after trip (breaker re-arm)
            cum = 0.0

    rate = trips / len(trade_pnls)
    limit = max_trip_rate
    if paper_mode:
        limit = min(limit, 0.35)

    if rate > limit:
        return _fail(
            f"KILL-SWITCH: trip_rate={rate:.2f} > max={limit:.2f} "
            f"({trips}/{len(trade_pnls)} trades) loss_limit={loss_limit} "
            f"paper_mode={paper_mode} - breaker too tight for evaluation "
            f"(farm: every paper fill tripped)",
            rule_count=len(trade_pnls),
            exploit_count=trips,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"kill switch ok trip_rate={rate:.2f} ({trips}/{len(trade_pnls)}) limit={loss_limit}"
        ),
        exit_code=0,
        rule_count=len(trade_pnls),
        exploit_count=trips,
        human_required=False,
    )

gate_price_book(bids, asks, *, order='best-first')

Gate bid/ask arrays for PRICE-BOOK-ORDER (Polymarket worst-to-best trap).

Farm failure: API returns worst-to-best; .first() took worst price.

Parameters:

Name	Type	Description	Default
bids	Sequence[float]	Bid prices (best bid should be highest).	required
asks	Sequence[float]	Ask prices (best ask should be lowest).	required
order	str	best-first expects best at index 0; worst-first documents the inverted API (still validates internal consistency).	'best-first'

Source code in src/balancelab/closed_loop.py

def gate_price_book(
    bids: Sequence[float],
    asks: Sequence[float],
    *,
    order: str = "best-first",
) -> GateOutcome:
    """Gate bid/ask arrays for PRICE-BOOK-ORDER (Polymarket worst-to-best trap).

    Farm failure: API returns worst-to-best; ``.first()`` took worst price.

    Args:
        bids: Bid prices (best bid should be highest).
        asks: Ask prices (best ask should be lowest).
        order: ``best-first`` expects best at index 0; ``worst-first`` documents
            the inverted API (still validates internal consistency).
    """
    if not bids or not asks:
        return _fail_loud(
            "empty price book - no bids or asks",
            rule_count=0,
        )

    best_bid = max(bids)
    best_ask = min(asks)
    if best_bid > best_ask:
        return _fail(
            f"PRICE-BOOK-ORDER: crossed book best_bid={best_bid} > best_ask={best_ask}",
            rule_count=len(bids) + len(asks),
        )

    if order == "best-first":
        if bids[0] != best_bid:
            return _fail(
                f"PRICE-BOOK-ORDER: expected best bid at [0], got {bids[0]} "
                f"(true best={best_bid}) - worst-first API trap "
                f"(farm: Polymarket .first())",
                rule_count=len(bids),
            )
        if asks[0] != best_ask:
            return _fail(
                f"PRICE-BOOK-ORDER: expected best ask at [0], got {asks[0]} "
                f"(true best={best_ask}) - worst-first API trap",
                rule_count=len(asks),
            )
    elif order == "worst-first":
        if bids[-1] != best_bid or asks[-1] != best_ask:
            return _fail(
                "PRICE-BOOK-ORDER: order=worst-first but last element is not best quote",
                rule_count=len(bids) + len(asks),
            )
    else:
        return _fail_loud(f"unknown price book order mode {order!r}")

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=f"price book ok order={order} best_bid={best_bid} best_ask={best_ask}",
        exit_code=0,
        rule_count=len(bids) + len(asks),
        human_required=False,
    )

recommend_fixes(report)

For each exploit found, suggest the minimum change to neutralize it.

Source code in src/balancelab/fixes.py

def recommend_fixes(report: ExploitReport) -> list[BalanceFix]:
    """For each exploit found, suggest the minimum change to neutralize it."""
    fixes: list[BalanceFix] = []

    for exploit in report.exploits:
        path = exploit.path
        gain_ratio = exploit.gain_ratio
        n_edges = max(1, len(path) - 1)

        # Determine target_edge: first edge in path if path has >= 2 elements
        target_edge: tuple[str, str] | None = None
        if len(path) >= 2:
            target_edge = (path[0], path[1])

        if len(path) == 2:
            # Single edge back and forth (2 nodes, one edge)
            fix_type = "daily_limit"
            suggested_value = None
            description = (
                f"Apply a daily transaction limit on the edge {path[0]} -> {path[1]} "
                f"to prevent repeated exploitation."
            )
            estimated_reduction_pct = 75.0

        elif gain_ratio > 2.0:
            fix_type = "rate_cap"
            # suggested_value: cap each edge rate so the cycle gain becomes 1.0
            # To neutralize: product of rates = 1.0
            # If current gain_ratio = R, cap each edge to (1/R)^(1/n_edges)
            suggested_value = (1.0 / max(gain_ratio, 1.001)) ** (1.0 / n_edges)
            description = (
                f"Cap exchange rate on edge ({path[0]} -> {path[1]}) "
                f"to {suggested_value:.4f} so the cycle gain reduces to 1.0."
            )
            estimated_reduction_pct = min(99.0, (gain_ratio - 1.0) / gain_ratio * 100)

        elif 1.0 <= gain_ratio <= 2.0:
            fix_type = "cooldown"
            suggested_value = gain_ratio * 10
            description = (
                f"Add a cooldown of {suggested_value:.1f} seconds between uses of this exploit "
                f"path to limit abuse (cycle gain: {gain_ratio:.2f}x)."
            )
            estimated_reduction_pct = 50.0

        else:
            fix_type = "require_prerequisite"
            suggested_value = None
            description = (
                f"Require a prerequisite item or condition before the path "
                f"{' -> '.join(path)} can be executed."
            )
            estimated_reduction_pct = 60.0

        fixes.append(
            BalanceFix(
                exploit_path=list(path),
                fix_type=fix_type,
                target_edge=target_edge,
                suggested_value=suggested_value,
                description=description,
                estimated_reduction_pct=estimated_reduction_pct,
            )
        )

    return fixes

critical_path(graph)

Find the sequence of nodes with highest economic throughput (most rules flow through).

Source code in src/balancelab/sensitivity.py

def critical_path(graph: EconomyGraph) -> list[str]:
    """Find the sequence of nodes with highest economic throughput (most rules flow through)."""
    if not graph.rules:
        return []

    # Build outgoing throughput per node: sum of exchange_rates for all rules where node is source
    throughput: dict[str, float] = {}
    for node in graph.items():
        throughput[node] = sum(r.exchange_rate() for r in graph.rules if r.source_item == node)

    # Start from the node with highest outgoing throughput
    if not throughput:
        return []

    start = max(throughput, key=lambda n: throughput[n])

    path: list[str] = [start]
    visited: set[str] = {start}
    current = start

    for _ in range(49):  # max 50 steps total (including start)
        neighbors = graph.neighbors(current)
        if not neighbors:
            break

        # Pick neighbor with highest exchange_rate rule
        best_target: str | None = None
        best_rate = -1.0
        for target, rule in neighbors:
            if rule.exchange_rate() > best_rate:
                best_rate = rule.exchange_rate()
                best_target = target

        if best_target is None or best_target in visited:
            break

        path.append(best_target)
        visited.add(best_target)
        current = best_target

    return path

sensitivity_analysis(graph, report)

Rank all nodes by how much they impact the economy balance, descending by impact_score.

Source code in src/balancelab/sensitivity.py

def sensitivity_analysis(graph: EconomyGraph, report: ExploitReport) -> list[SensitivityResult]:
    """Rank all nodes by how much they impact the economy balance, descending by impact_score."""
    total_rules = max(1, len(graph.rules))
    results: list[SensitivityResult] = []

    for node in graph.items():
        # Count how many rules reference this node as source vs target
        as_source = sum(1 for r in graph.rules if r.source_item == node)
        as_target = sum(1 for r in graph.rules if r.target_item == node)
        connected_rules = as_source + as_target

        # Determine node_type
        if as_source > 0 and as_target > 0:
            node_type = "hub"
        elif as_source > 0:
            node_type = "source_only"
        else:
            node_type = "target_only"

        # Count exploit involvement
        exploit_involvement = sum(1 for exploit in report.exploits if node in exploit.path)

        # Compute impact_score
        impact_score = min(
            1.0,
            (connected_rules * 0.3 + exploit_involvement * 0.5) / total_rules,
        )

        # Recommendation
        if exploit_involvement >= 2:
            recommendation = "gate"
        elif exploit_involvement >= 1:
            recommendation = "rate-limit"
        else:
            recommendation = "monitor"

        results.append(
            SensitivityResult(
                node_id=node,
                node_type=node_type,
                impact_score=impact_score,
                connected_rules=connected_rules,
                exploit_involvement=exploit_involvement,
                recommendation=recommendation,
            )
        )

    # Sort by impact_score descending
    results.sort(key=lambda r: r.impact_score, reverse=True)
    return results

simulate(graph, initial_levels, n_steps=100, agent_strategy='greedy')

Run economy simulation. 'exploit' strategy finds and uses exploits.

Source code in src/balancelab/simulation.py

def simulate(
    graph: EconomyGraph,
    initial_levels: dict[str, float],
    n_steps: int = 100,
    agent_strategy: str = "greedy",  # "greedy" | "balanced" | "exploit"
) -> SimulationResult:
    """Run economy simulation. 'exploit' strategy finds and uses exploits."""
    resource_levels: dict[str, float] = dict(initial_levels)
    steps: list[SimulationStep] = []
    all_violated: set[str] = set()
    inflation_detected = False
    inflation_resource: str | None = None

    # Pre-compute exploit paths if using exploit strategy
    exploit_cycles: list[list[str]] = []
    if agent_strategy == "exploit":
        finder = ExploitFinder()
        report = finder.find_exploits(graph)
        for exploit in report.exploits:
            exploit_cycles.append(exploit.path)

    for step_num in range(1, n_steps + 1):
        activity_counts: dict[str, int] = {}
        rule_violations: list[str] = []

        if agent_strategy == "greedy":
            for rule in graph.rules:
                if resource_levels.get(rule.source_item, 0.0) >= rule.source_qty:
                    resource_levels[rule.source_item] = (
                        resource_levels.get(rule.source_item, 0.0) - rule.source_qty
                    )
                    resource_levels[rule.target_item] = (
                        resource_levels.get(rule.target_item, 0.0) + rule.target_qty
                    )
                    activity_key = f"{rule.source_item}->{rule.target_item}"
                    activity_counts[activity_key] = activity_counts.get(activity_key, 0) + 1
                else:
                    rule_violations.append(rule.id)

        elif agent_strategy == "balanced":
            # At most 1 rule per source item per step - pick the best exchange rate
            best_rules: dict[str, EconomyRule] = {}
            for rule in graph.rules:
                src = rule.source_item
                if src not in best_rules or rule.exchange_rate() > best_rules[src].exchange_rate():
                    best_rules[src] = rule

            for _src, rule in best_rules.items():
                if resource_levels.get(rule.source_item, 0.0) >= rule.source_qty:
                    resource_levels[rule.source_item] = (
                        resource_levels.get(rule.source_item, 0.0) - rule.source_qty
                    )
                    resource_levels[rule.target_item] = (
                        resource_levels.get(rule.target_item, 0.0) + rule.target_qty
                    )
                    activity_key = f"{rule.source_item}->{rule.target_item}"
                    activity_counts[activity_key] = activity_counts.get(activity_key, 0) + 1
                else:
                    rule_violations.append(rule.id)

        elif agent_strategy == "exploit":
            # Try to execute each exploit cycle once per step
            for cycle_path in exploit_cycles:
                if len(cycle_path) < 2:
                    continue
                # Find rules that form the cycle path
                can_execute = True
                cycle_rules: list[EconomyRule] = []
                for i in range(len(cycle_path) - 1):
                    src = cycle_path[i]
                    tgt = cycle_path[i + 1]
                    rule_found = None
                    for rule in graph.rules:
                        if rule.source_item == src and rule.target_item == tgt:
                            rule_found = rule
                            break
                    if rule_found is None or resource_levels.get(src, 0.0) < rule_found.source_qty:
                        can_execute = False
                        if rule_found is not None:
                            rule_violations.append(rule_found.id)
                        break
                    cycle_rules.append(rule_found)

                if can_execute:
                    for rule in cycle_rules:
                        resource_levels[rule.source_item] = (
                            resource_levels.get(rule.source_item, 0.0) - rule.source_qty
                        )
                        resource_levels[rule.target_item] = (
                            resource_levels.get(rule.target_item, 0.0) + rule.target_qty
                        )
                        activity_key = f"{rule.source_item}->{rule.target_item}"
                        activity_counts[activity_key] = activity_counts.get(activity_key, 0) + 1

            # Skip greedy pass - exploit cycles were already applied
            # (applying greedy rules again would double-count rules used in exploit cycles)

        # Check for inflation: any resource > 10x its initial level
        if not inflation_detected:
            for item, level in resource_levels.items():
                init = initial_levels.get(item, 0.0)
                if init > 0 and level > 10.0 * init:
                    inflation_detected = True
                    inflation_resource = item
                    break

        # Track all violations
        all_violated.update(rule_violations)

        steps.append(
            SimulationStep(
                step=step_num,
                resource_levels=dict(resource_levels),
                activity_counts=dict(activity_counts),
                rule_violations=list(rule_violations),
            )
        )

    final_levels = dict(resource_levels)
    violated_rules = sorted(all_violated)

    inflation_str = "yes" if inflation_detected else "no"
    summary = f"Ran {n_steps} steps. Final levels: {final_levels}. Inflation: {inflation_str}."

    return SimulationResult(
        steps=steps,
        final_levels=final_levels,
        violated_rules=violated_rules,
        inflation_detected=inflation_detected,
        inflation_resource=inflation_resource,
        summary=summary,
    )