Account Adjustments

Account adjustment models non-trade operations (NTO): direct state corrections that do not originate from order execution.

Typical examples:

initial account load;
balance top-up or withdrawal;
position correction after a corporate action;
settlement or funding transfer represented as an explicit adjustment batch.

Operation Types

Account adjustment balance operation: physical asset balance correction for one asset.
Account adjustment position operation: derivatives-like position correction for one instrument + collateral asset.
Account PnL operation: replaces the single account-wide PnL state with a numeric PnL or an explicit Halted reason. It carries no asset.
A balance operation may also replace the realized-PnL state for its asset slot with a numeric PnL or an explicit Halted reason.

Balance and position operations may include optional amount and bounds groups:

Account adjustment amount: balance, held, incoming.
Account adjustment bounds: inclusive lower/upper constraints for each amount field.

Account PnL is separate from balance amounts and average-entry-price corrections. Account and position PnL states re-arm independently: an account PnL operation never changes a position state, and an asset-scoped balance PnL correction never changes the account-wide PnL used by the Spot Funds kill switch.

Batch Semantics

Apply account adjustments validates the input as an atomic batch:

evaluation order is deterministic: per-adjustment slice order, then per-policy registration order;
validation stops at the first reject;
outcome is all-or-nothing - a reject means the full batch is rejected;
on reject, the caller receives the failed index together with the policy reject payload.

Operational guidance:

If external logic needs to correct policy-related state while the engine is active, prefer apply account adjustments over direct parallel mutation of custom-policy fields.
Direct parallel access to custom-policy state still requires host-side synchronization.

Outcomes

A successful balance or position batch returns one outcome per asset each adjustment touched. Amount outcomes carry a signed delta and an absolute snapshot. A balance PnL correction publishes its numeric result or halt reason as an asset-scoped adjustment outcome. An account-PnL operation only replaces the account-wide state and produces no per-asset adjustment outcome. Use amount delta values to keep your own ledger aligned with the engine; treat absolute as diagnostic only. See Balance Reconciliation for the full delta-versus-absolute contract and a scheme for persisting limits in your own store. The accepted batch result also reports any account blocks that a policy produced while accepting the adjustment. The engine records those blocks before it returns the result, so a reported block means later pre-trade starts for that account are already rejected.

Examples

Go

accountID := param.NewAccountIDFromUint64(99224416)

usd, err := param.NewAsset("USD")
if err != nil {
  panic(err)
}
spx, err := param.NewAsset("SPX")
if err != nil {
  panic(err)
}
entryPrice, _ := param.NewPriceFromString("95000")
totalCash, _ := param.NewPositionSizeFromString("10000")
totalPosition, _ := param.NewPositionSizeFromString("-3")

cashAdj, err := model.NewAccountAdjustmentFromValues(model.AccountAdjustmentValues{
  BalanceOperation: optional.Some(
    model.NewAccountAdjustmentBalanceOperationFromValues(
      model.AccountAdjustmentBalanceOperationValues{
        Asset: optional.Some(usd),
      },
    ),
  ),
  Amount: optional.Some(
    model.NewAccountAdjustmentAmountFromValues(model.AccountAdjustmentAmountValues{
      Balance: optional.Some(param.NewAbsoluteAdjustmentAmount(totalCash)),
    }),
  ),
})
if err != nil {
  panic(err)
}

posAdj, err := model.NewAccountAdjustmentFromValues(model.AccountAdjustmentValues{
  PositionOperation: optional.Some(
    model.NewAccountAdjustmentPositionOperationFromValues(
      model.AccountAdjustmentPositionOperationValues{
        Instrument:        optional.Some(param.NewInstrument(spx, usd)),
        CollateralAsset:   optional.Some(usd),
        AverageEntryPrice: optional.Some(entryPrice),
        Mode:              optional.Some(param.PositionModeHedged),
      },
    ),
  ),
  Amount: optional.Some(
    model.NewAccountAdjustmentAmountFromValues(model.AccountAdjustmentAmountValues{
      Balance: optional.Some(param.NewAbsoluteAdjustmentAmount(totalPosition)),
    }),
  ),
})
if err != nil {
  panic(err)
}

// On accept, BatchError is unset and Outcomes holds the per-asset adjustment
// outcomes. AccountBlocks reports blocks that were recorded before return.
result, err := engine.ApplyAccountAdjustment(
  accountID,
  []model.AccountAdjustment{cashAdj, posAdj},
)
if err != nil {
  panic(err)
}
if result.BatchError.IsSet() {
  panic("unexpected account-adjustment reject")
}
if len(result.AccountBlocks) != 0 {
  panic("unexpected account block")
}

Python

import openpit
import openpit.pretrade.policies

# Build one batch that mixes balance and position adjustments.
account_id = openpit.param.AccountId.from_int(99224416)

adjustments = [
    openpit.AccountAdjustment(
        operation=openpit.AccountAdjustmentBalanceOperation(asset="USD"),
        amount=openpit.AccountAdjustmentAmount(
            balance=openpit.param.AdjustmentAmount.absolute(
                openpit.param.PositionSize(10000)
            )
        ),
    ),
    openpit.AccountAdjustment(
        operation=openpit.AccountAdjustmentPositionOperation(
            instrument=openpit.Instrument("SPX", "USD"),
            collateral_asset="USD",
            average_entry_price=openpit.param.Price(95000),
            mode=openpit.param.PositionMode.HEDGED,
        ),
        amount=openpit.AccountAdjustmentAmount(
            balance=openpit.param.AdjustmentAmount.absolute(
                openpit.param.PositionSize(-3)
            )
        ),
    ),
]

# The engine validates the whole batch atomically.
engine = (
    openpit.Engine.builder()
    .no_sync()
    .builtin(openpit.pretrade.policies.build_order_validation())
    .build()
)
result = engine.apply_account_adjustment(
    account_id=account_id,
    adjustments=adjustments,
)
assert result.ok
assert not result.account_blocks

JavaScript

import { Engine } from "@openpit/engine";
import { type AccountAdjustmentInit } from "@openpit/engine/model";
import { AdjustmentAmount } from "@openpit/engine/param";
import { buildOrderValidation } from "@openpit/engine/pretrade/policies";

// Build one batch that mixes balance and position adjustments. Each adjustment
// is a plain object literal; position sizes cross the boundary as decimal
// strings.
const accountId = 99224416;

const adjustments: AccountAdjustmentInit[] = [
  {
    operation: { asset: "USD" },
    amount: { balance: AdjustmentAmount.absolute("10000") },
  },
  {
    operation: {
      underlyingAsset: "SPX",
      settlementAsset: "USD",
      collateralAsset: "USD",
      averageEntryPrice: "95000",
      mode: "hedged",
    },
    amount: { balance: AdjustmentAmount.absolute("-3") },
  },
];

// The engine validates the whole batch atomically.
const engine = Engine.builder()
  .builtin(buildOrderValidation())
  .build();
const result = engine.applyAccountAdjustment(accountId, adjustments);
if (!result.ok) {
  throw new Error("account-adjustment batch must be accepted");
}
if (result.accountBlocks.length !== 0) {
  throw new Error("unexpected account block");
}

C++

#include <openpit/openpit.hpp>

#include <cassert>

namespace aa = openpit::accountadjustment;
namespace param = openpit::param;
namespace policies = openpit::pretrade::policies;

// Build one batch that mixes balance and position adjustments.
const param::AccountId accountId = param::AccountId::FromUint64(99224416);

aa::AccountAdjustment cashAdj;
{
  aa::BalanceOperation balance;
  balance.asset = ::openpit::param::Asset("USD");
  cashAdj.operation = aa::Operation::OfBalance(std::move(balance));
  aa::Amount amount;
  amount.balance =
      param::AdjustmentAmount::Absolute(param::PositionSize::FromString("10000"));
  cashAdj.amount = std::move(amount);
}

aa::AccountAdjustment posAdj;
{
  aa::PositionOperation position;
  position.instrument = openpit::model::Instrument(
      ::openpit::param::Asset("SPX"), ::openpit::param::Asset("USD"));
  position.collateralAsset = ::openpit::param::Asset("USD");
  position.averageEntryPrice = param::Price::FromString("95000");
  position.mode = openpit::model::PositionMode::Hedged;
  posAdj.operation = aa::Operation::OfPosition(std::move(position));
  aa::Amount amount;
  amount.balance =
      param::AdjustmentAmount::Absolute(param::PositionSize::FromString("-3"));
  posAdj.amount = std::move(amount);
}

const std::vector<aa::AccountAdjustment> adjustments{std::move(cashAdj),
                                                     std::move(posAdj)};

// The engine validates the whole batch atomically.
openpit::EngineBuilder builder(openpit::SyncPolicy::None);
builder.Add(policies::OrderValidationPolicy{});
const openpit::Engine engine = builder.Build();

// On accept the result passes and carries the per-asset account-adjustment
// outcomes.
const openpit::AdjustmentResult result =
    engine.ApplyAccountAdjustment(accountId, adjustments);
assert(result.Passed());
assert(result.accountBlocks.empty());

Rust

use openpit::param::{
    AccountId, AdjustmentAmount, Asset, PositionMode, PositionSize, Price,
};
use openpit::{
    AccountAdjustmentAmount, AccountAdjustmentBalanceOperation,
    AccountAdjustmentPositionOperation, Engine, Instrument,
};

#[derive(Clone)]
enum AccountAdjustmentOperation {
    Balance(AccountAdjustmentBalanceOperation),
    Position(AccountAdjustmentPositionOperation),
}

#[derive(Clone)]
struct AccountAdjustment {
    operation: AccountAdjustmentOperation,
    amount: AccountAdjustmentAmount,
}

// Build one batch that mixes balance and position adjustments.
let account_id = AccountId::from_u64(99224416);

let adjustments = vec![
    AccountAdjustment {
        operation: AccountAdjustmentOperation::Balance(
            AccountAdjustmentBalanceOperation {
                asset: Asset::new("USD")?,
                average_entry_price: None,
            },
        ),
        amount: AccountAdjustmentAmount {
            balance: Some(AdjustmentAmount::Absolute(
                PositionSize::from_f64(10000.0)?,
            )),
            held: None,
            incoming: None,
        },
    },
    AccountAdjustment {
        operation: AccountAdjustmentOperation::Position(
            AccountAdjustmentPositionOperation {
                instrument: Instrument::new(
                    Asset::new("SPX")?,
                    Asset::new("USD")?,
                ),
                collateral_asset: Asset::new("USD")?,
                average_entry_price: Price::from_f64(95000.0)?,
                mode: PositionMode::Hedged,
                leverage: None,
            },
        ),
        amount: AccountAdjustmentAmount {
            balance: Some(AdjustmentAmount::Absolute(
                PositionSize::from_f64(-3.0)?,
            )),
            held: None,
            incoming: None,
        },
    },
];

struct AcceptAllAdjustments;

impl<Sync> openpit::pretrade::PreTradePolicy<(), (), AccountAdjustment, Sync>
    for AcceptAllAdjustments
where
    Sync: openpit::SyncMode,
{
    fn name(&self) -> &'static str {
        "AcceptAllAdjustments"
    }

    fn apply_account_adjustment(
        &self,
        _ctx: &openpit::AccountAdjustmentContext<
            <Sync as openpit::SyncMode>::StorageLockingPolicyFactory,
        >,
        _account_id: openpit::param::AccountId,
        _adjustment: &AccountAdjustment,
        _mutations: &mut openpit::Mutations,
    ) -> Result<openpit::pretrade::PolicyAccountAdjustmentResult, openpit::pretrade::Rejects> {
        Ok(openpit::pretrade::PolicyAccountAdjustmentResult {
            account_adjustments: Vec::new(),
            account_blocks: Vec::new(),
        })
    }
}

// The engine validates the whole batch atomically.
let engine = Engine::builder::<(), (), AccountAdjustment>()
    .no_sync()
    .pre_trade(AcceptAllAdjustments)
    .build()?;
let result = engine.apply_account_adjustment(account_id, &adjustments)?;
assert!(result.account_blocks.is_empty());

Writing an Account Adjustment Policy with Rollback

Account adjustment policies can register rollback actions that undo intermediate state if a later element in the batch is rejected. The engine applies rollback actions in reverse registration order (last registered = first rolled back).

Rollback Safety

Account adjustment batches run within a single engine call. No external system (venue, risk aggregator) observes intermediate state between elements, so rollback by absolute value is safe: a policy can capture the current value before modification and restore it on rollback without risking inconsistency.

The pre-trade pipeline is different: a reservation may be observed by external systems between creation and finalization, so pre-trade policies should prefer delta-based rollback.

Example: Balance Limit Policy

The policy below tracks cumulative adjustment totals per asset and rejects the batch if any asset exceeds a configured limit. On rejection, all previously accumulated totals are rolled back to their state before the batch started.

Go

type CumulativeLimitPolicy struct {
  maxCumulative param.PositionSize
  totals        map[string]param.PositionSize
}

func (*CumulativeLimitPolicy) Close() {}

func (*CumulativeLimitPolicy) Name() string { return "CumulativeLimitPolicy" }

func (*CumulativeLimitPolicy) PolicyGroupID() model.PolicyGroupID {
    return model.DefaultPolicyGroupID
}

func (*CumulativeLimitPolicy) CheckPreTradeStart(
  pretrade.Context,
  model.Order,
) []reject.Reject {
  return nil
}

func (*CumulativeLimitPolicy) PerformPreTradeCheck(
  pretrade.Context,
  model.Order,
  tx.Mutations,
  pretrade.Result,
) []reject.Reject {
  return nil
}

func (*CumulativeLimitPolicy) ApplyExecutionReport(
  pretrade.PostTradeContext,
  model.ExecutionReport,
  pretrade.PostTradeAdjustments,
  pretrade.PostTradePnls,
) []reject.AccountBlock {
  return nil
}

func (v *CumulativeLimitPolicy) ApplyAccountAdjustment(
  _ accountadjustment.Context,
  _ param.AccountID,
  adjustment model.AccountAdjustment,
  mutations tx.Mutations,
  _ pretrade.AccountOutcomes,
) (pretrade.PolicyAccountAdjustmentResult, []reject.Reject) {
  operation, ok := adjustment.BalanceOperation().Get()
  if !ok {
    return pretrade.PolicyAccountAdjustmentResult{}, nil
  }
  asset, ok := operation.Asset().Get()
  if !ok {
    return pretrade.PolicyAccountAdjustmentResult{}, nil
  }
  amounts, ok := adjustment.Amount().Get()
  if !ok {
    return pretrade.PolicyAccountAdjustmentResult{}, nil
  }
  balance, ok := amounts.Balance().Get()
  if !ok {
    return pretrade.PolicyAccountAdjustmentResult{}, nil
  }

  if v.totals == nil {
    v.totals = make(map[string]param.PositionSize)
  }

  assetID := asset.String()
  previous, existed := v.totals[assetID]
  current := previous
  if !existed {
    current = param.NewPositionSizeZero()
  }

  var next param.PositionSize
  if balance.IsAbsolute() {
    next = balance.MustAbsolute()
  } else {
    var err error
    next, err = current.CheckedAdd(balance.MustDelta())
    if err != nil {
      return pretrade.PolicyAccountAdjustmentResult{}, reject.NewSingleItemList(
        reject.CodeArithmeticOverflow,
        v.Name(),
        "invalid adjustment total",
        err.Error(),
        reject.ScopeAccount,
      )
    }
  }

  if next.Compare(v.maxCumulative) > 0 {
    return pretrade.PolicyAccountAdjustmentResult{}, reject.NewSingleItemList(
      reject.CodeRiskLimitExceeded,
      v.Name(),
      "cumulative limit exceeded",
      fmt.Sprintf("%s: %s > %s", assetID, next, v.maxCumulative),
      reject.ScopeAccount,
    )
  }

  v.totals[assetID] = next
  rollback := func() {
    if existed {
      v.totals[assetID] = previous
    } else {
      delete(v.totals, assetID)
    }
  }
  if err := mutations.Push(func() {}, rollback); err != nil {
    rollback()
    return pretrade.PolicyAccountAdjustmentResult{}, reject.NewSingleItemList(
      reject.CodeSystemUnavailable,
      v.Name(),
      "failed to register adjustment rollback",
      err.Error(),
      reject.ScopeAccount,
    )
  }
  return pretrade.PolicyAccountAdjustmentResult{}, nil
}

Python

import openpit


class CumulativeLimitPolicy(openpit.pretrade.Policy):
    """Tracks cumulative totals per asset, rejects batch on limit breach."""

    def __init__(self, max_cumulative: openpit.param.PositionSize) -> None:
        self._max = max_cumulative
        self._totals: dict[str, openpit.param.PositionSize] = {}

    @property
    def name(self) -> str:
        return "CumulativeLimitPolicy"

    def apply_account_adjustment(
        self,
        ctx: openpit.AccountAdjustmentContext,
        account_id: openpit.param.AccountId,
        adjustment: openpit.AccountAdjustment,
    ) -> openpit.pretrade.PolicyAccountAdjustmentResult:
        del ctx, account_id
        operation = adjustment.operation
        if not isinstance(operation, openpit.AccountAdjustmentBalanceOperation):
            return openpit.pretrade.PolicyAccountAdjustmentResult()

        asset_id = operation.asset
        amount = adjustment.amount
        balance = amount.balance if amount is not None else None
        if asset_id is None or balance is None:
            return openpit.pretrade.PolicyAccountAdjustmentResult()

        previous = self._totals.get(asset_id)
        current = previous if previous is not None else openpit.param.PositionSize("0")
        absolute = balance.as_absolute
        if absolute is not None:
            new_total = absolute
        else:
            delta = balance.as_delta
            if delta is None:
                return openpit.pretrade.PolicyAccountAdjustmentResult()
            new_total = current + delta

        # Reject if the limit is breached.
        if new_total > self._max:
            return openpit.pretrade.PolicyAccountAdjustmentResult(
                rejects=(
                    openpit.pretrade.PolicyReject(
                        code=openpit.pretrade.RejectCode.RISK_LIMIT_EXCEEDED,
                        reason="cumulative limit exceeded",
                        details=f"{asset_id}: {new_total} > {self._max}",
                        scope=openpit.pretrade.RejectScope.ACCOUNT,
                    ),
                ),
            )

        # Apply immediately so later adjustments in the same batch see it.
        self._totals[asset_id] = new_total

        def rollback() -> None:
            if previous is None:
                self._totals.pop(asset_id, None)
            else:
                self._totals[asset_id] = previous

        return openpit.pretrade.PolicyAccountAdjustmentResult(
            mutations=(
                openpit.Mutation(
                    commit=lambda: None,
                    rollback=rollback,
                ),
            ),
        )


def balance_adjustment(
    amount: openpit.param.AdjustmentAmount,
) -> openpit.AccountAdjustment:
    return openpit.AccountAdjustment(
        operation=openpit.AccountAdjustmentBalanceOperation(asset="USD"),
        amount=openpit.AccountAdjustmentAmount(balance=amount),
    )


policy = CumulativeLimitPolicy(openpit.param.PositionSize("100"))
engine = openpit.Engine.builder().no_sync().pre_trade(policy=policy).build()
account_id = openpit.param.AccountId.from_int(99224416)

seed = engine.apply_account_adjustment(
    account_id=account_id,
    adjustments=[
        balance_adjustment(
            openpit.param.AdjustmentAmount.absolute(
                openpit.param.PositionSize("40")
            )
        )
    ],
)
rejected = engine.apply_account_adjustment(
    account_id=account_id,
    adjustments=[
        balance_adjustment(
            openpit.param.AdjustmentAmount.delta(openpit.param.PositionSize("30"))
        ),
        balance_adjustment(
            openpit.param.AdjustmentAmount.delta(openpit.param.PositionSize("40"))
        ),
    ],
)
after_rollback = engine.apply_account_adjustment(
    account_id=account_id,
    adjustments=[
        balance_adjustment(
            openpit.param.AdjustmentAmount.delta(openpit.param.PositionSize("60"))
        )
    ],
)

assert seed.ok
assert not rejected.ok
assert rejected.failed_index == 1
assert rejected.rejects[0].code == openpit.pretrade.RejectCode.RISK_LIMIT_EXCEEDED
assert after_rollback.ok

JavaScript

import { Engine } from "@openpit/engine";
import { type AccountAdjustmentContext } from "@openpit/engine/accountadjustment";
import { type AccountAdjustment } from "@openpit/engine/model";
import {
  AccountId,
  AdjustmentAmount,
  PositionSize,
} from "@openpit/engine/param";
import {
  type Policy,
  type PolicyAccountAdjustmentResult,
  type PolicyPreTradeResult,
  type PolicyReject,
} from "@openpit/engine/pretrade";

// Tracks cumulative totals per asset, rejects the batch on a limit breach.
class CumulativeLimitPolicy implements Policy {
  readonly name = "CumulativeLimitPolicy";
  private readonly totals = new Map<string, PositionSize>();

  constructor(private readonly maxCumulative: PositionSize) {}

  // The pre-trade hooks are required by the Policy interface; this policy only
  // acts on the account-adjustment path, so they accept with no contribution.
  checkPreTradeStart(): Iterable<PolicyReject> {
    return [];
  }

  performPreTradeCheck(): PolicyPreTradeResult {
    return {};
  }

  applyAccountAdjustment(
    _ctx: AccountAdjustmentContext,
    _accountId: AccountId,
    adjustment: AccountAdjustment,
  ): PolicyAccountAdjustmentResult {
    // Use the asset as the aggregation key for the cumulative limit.
    const operation = adjustment.operation;
    const assetId =
      operation !== undefined && "asset" in operation
        ? (operation.asset ?? "")
        : "";

    const previous = this.totals.get(assetId);
    const current = previous ?? PositionSize.fromInt(0n);
    const balance = adjustment.amount?.balance;
    if (balance === undefined) {
      return { accountBlocks: [] };
    }

    const absolute = balance.asAbsolute;
    let newTotal: PositionSize;
    if (absolute !== undefined) {
      newTotal = absolute;
    } else {
      const delta = balance.asDelta;
      if (delta === undefined) {
        return { accountBlocks: [] };
      }
      newTotal = current.add(delta);
    }

    // Reject if the limit is breached.
    if (newTotal.compare(this.maxCumulative) > 0) {
      return {
        rejects: [{
          code: "RiskLimitExceeded",
          reason: "cumulative limit exceeded",
          details: `${assetId}: ${newTotal.toString()} > ${this.maxCumulative.toString()}`,
          scope: "account",
        }],
        accountBlocks: [],
      };
    }

    // Apply immediately so later adjustments in the same batch see the updated
    // total.
    this.totals.set(assetId, newTotal);

    // Rollback by absolute value - safe in the account-adjustment pipeline
    // because no external system sees intermediate batch state. Commit is empty:
    // the state was applied eagerly.
    return {
      mutations: [{
        commit: () => {},
        rollback: () => {
          if (previous === undefined) {
            this.totals.delete(assetId);
          } else {
            this.totals.set(assetId, previous);
          }
        },
      }],
      accountBlocks: [],
    };
  }
}

// Seed an absolute value, then reject a batch after its first delta was
// applied. A final delta reaches the limit exactly only if the failed batch
// rolled its first mutation back.
const engine = Engine.builder()
  .preTrade(new CumulativeLimitPolicy(PositionSize.fromString("100")))
  .build();
const accountId = 99224416;
const adjustment = (
  amount: ReturnType<typeof AdjustmentAmount.absolute>,
) => ({
  operation: { asset: "USD" },
  amount: { balance: amount },
});

const seed = engine.applyAccountAdjustment(accountId, [
  adjustment(AdjustmentAmount.absolute("40")),
]);

const rejected = engine.applyAccountAdjustment(accountId, [
  adjustment(AdjustmentAmount.delta("30")),
  adjustment(AdjustmentAmount.delta("40")),
]);

const afterRollback = engine.applyAccountAdjustment(accountId, [
  adjustment(AdjustmentAmount.delta("60")),
]);

if (!seed.ok) {
  throw new Error("the absolute seed must be accepted");
}
if (
  rejected.ok ||
  rejected.failedIndex !== 1 ||
  rejected.rejects[0]?.code !== "RiskLimitExceeded"
) {
  throw new Error("the second delta must breach the cumulative limit");
}
if (!afterRollback.ok) {
  throw new Error("the failed batch must roll the first delta back");
}

C++

The C++ account-adjustment callback receives the current adjustment, a mutation collector, and an account-outcome collector. Apply policy-local state eagerly so later elements in the same batch see it, then register commit and rollback callbacks with tx::Mutations::Push. If any policy rejects an element, OpenPit runs the preceding rollbacks in reverse order and rejects the whole batch.

#include <openpit/openpit.hpp>

#include <map>
#include <memory>
#include <optional>
#include <string>
#include <string_view>

struct CumulativeLimitState {
  std::map<std::string, openpit::param::PositionSize> totals;
  std::size_t commits = 0;
  std::size_t rollbacks = 0;
};

// Tracks the last accepted absolute balance per asset. Each accepted element
// mutates eagerly so the next element in the same batch sees it, then registers
// commit/rollback callbacks with the engine transaction.
class CumulativeLimitPolicy {
 public:
  CumulativeLimitPolicy(
      openpit::param::PositionSize maxCumulative,
      std::shared_ptr<CumulativeLimitState> state)
      : m_maxCumulative(maxCumulative), m_state(std::move(state)) {}

  [[nodiscard]] std::string_view Name() const noexcept {
    return "CumulativeLimitPolicy";
  }

  [[nodiscard]] openpit::pretrade::PolicyAccountAdjustmentResult
  ApplyAccountAdjustment(
      const openpit::accountadjustment::Context& context,
      openpit::param::AccountId accountId,
      const openpit::accountadjustment::AccountAdjustment& adjustment,
      openpit::tx::Mutations& mutations,
      openpit::pretrade::AccountOutcomes& outcomes) const {
    static_cast<void>(context);
    static_cast<void>(accountId);
    static_cast<void>(outcomes);

    const auto* balance = adjustment.operation
                              ? adjustment.operation->AsBalance()
                              : nullptr;
    const auto absolute = adjustment.amount && adjustment.amount->balance
                              ? adjustment.amount->balance->AsAbsolute()
                              : std::nullopt;
    if (balance == nullptr || !balance->asset || !adjustment.amount ||
        !adjustment.amount->balance || !absolute) {
      return {};
    }

    const std::string asset(balance->asset->View());
    const openpit::param::PositionSize next =
        *absolute;
    if (next > m_maxCumulative) {
      openpit::pretrade::PolicyAccountAdjustmentResult result;
      result.decision.Push(openpit::pretrade::Reject(
          std::string(Name()), openpit::pretrade::RejectScope::Account,
          openpit::pretrade::RejectCode::RiskLimitExceeded,
          "cumulative limit exceeded",
          asset + " absolute balance exceeds the configured limit"));
      return result;
    }

    std::optional<openpit::param::PositionSize> previous;
    if (const auto it = m_state->totals.find(asset);
        it != m_state->totals.end()) {
      previous = it->second;
    }
    m_state->totals.insert_or_assign(asset, next);

    const auto state = m_state;
    mutations.Push(
        [state] { ++state->commits; },
        [state, asset, previous] {
          ++state->rollbacks;
          if (previous) {
            state->totals.insert_or_assign(asset, *previous);
          } else {
            state->totals.erase(asset);
          }
        });
    return {};
  }

 private:
  openpit::param::PositionSize m_maxCumulative;
  std::shared_ptr<CumulativeLimitState> m_state;
};

[[nodiscard]] openpit::accountadjustment::AccountAdjustment AbsoluteBalance(
    openpit::param::Asset asset, std::string_view value) {
  openpit::accountadjustment::BalanceOperation balance;
  balance.asset = std::move(asset);

  openpit::accountadjustment::Amount amount;
  amount.balance = openpit::param::AdjustmentAmount::Absolute(
      openpit::param::PositionSize::FromString(value));

  openpit::accountadjustment::AccountAdjustment adjustment;
  adjustment.operation =
      openpit::accountadjustment::Operation::OfBalance(std::move(balance));
  adjustment.amount = std::move(amount);
  return adjustment;
}

const auto state = std::make_shared<CumulativeLimitState>();
openpit::EngineBuilder builder(openpit::SyncPolicy::None);
openpit::pretrade::CustomPolicy<CumulativeLimitPolicy> policy(
    "CumulativeLimitPolicy",
    CumulativeLimitPolicy(openpit::param::PositionSize::FromString("1000000"),
                          state));
builder.Add(policy);
const openpit::Engine engine = builder.Build();

const openpit::param::AccountId accountId =
    openpit::param::AccountId::FromUint64(99224416);

const openpit::AdjustmentResult accepted = engine.ApplyAccountAdjustment(
    accountId, std::vector<openpit::accountadjustment::AccountAdjustment>{
                   AbsoluteBalance(openpit::param::Asset("USD"), "100")});
assert(accepted.Passed());
assert(state->totals.at("USD").ToString() == "100");
assert(state->commits == 1);

const openpit::AdjustmentResult rejected = engine.ApplyAccountAdjustment(
    accountId, std::vector<openpit::accountadjustment::AccountAdjustment>{
                   AbsoluteBalance(openpit::param::Asset("USD"), "200"),
                   AbsoluteBalance(openpit::param::Asset("USD"), "2000000")});
assert(!rejected.Passed());
assert(rejected.batchError->FailedAdjustmentIndex() == 1);
assert(state->totals.at("USD").ToString() == "100");
assert(state->commits == 1);
assert(state->rollbacks == 1);

Rust

use std::sync::Arc;

use openpit::param::{AccountId, Volume};
use openpit::pretrade::{Reject, RejectCode, RejectScope, Rejects};
use openpit::storage::{
    CreateStorageFor, LockingPolicyFactory, Storage, StorageBuilder,
};
use openpit::pretrade::PreTradePolicy;
use openpit::{AccountAdjustmentContext, Mutation, Mutations};

struct BalanceLimitPolicy<StorageLockingPolicyFactory>
where
    StorageLockingPolicyFactory: LockingPolicyFactory,
{
    max_total: Volume,
    totals: Arc<
        Storage<
            String,
            Volume,
            StorageLockingPolicyFactory::Policy,
        >,
    >,
}

impl<StorageLockingPolicyFactory>
    BalanceLimitPolicy<StorageLockingPolicyFactory>
where
    StorageLockingPolicyFactory: LockingPolicyFactory
        + CreateStorageFor<String>,
{
    fn new(
        max_total: Volume,
        storage_builder: &StorageBuilder<StorageLockingPolicyFactory>,
    ) -> Self {
        Self {
            max_total,
            totals: Arc::new(storage_builder.create_for_bound_key()),
        }
    }
}

/// Adjustment type must expose an asset and a delta amount.
trait HasAssetDelta {
    fn asset_id(&self) -> &str;
    fn delta(&self) -> Volume;
}

impl<Order, ExecutionReport, A, Sync, StorageLockingPolicyFactory>
    PreTradePolicy<Order, ExecutionReport, A, Sync>
    for BalanceLimitPolicy<StorageLockingPolicyFactory>
where
    A: HasAssetDelta,
    Sync: openpit::SyncMode,
    StorageLockingPolicyFactory: LockingPolicyFactory
        + CreateStorageFor<String>,
    StorageLockingPolicyFactory::Policy: 'static,
{
    fn name(&self) -> &str {
        "BalanceLimitPolicy"
    }

    fn apply_account_adjustment(
        &self,
        _ctx: &AccountAdjustmentContext<
            <Sync as openpit::SyncMode>::StorageLockingPolicyFactory,
        >,
        _account_id: AccountId,
        adjustment: &A,
        mutations: &mut Mutations,
    ) -> Result<openpit::pretrade::PolicyAccountAdjustmentResult, Rejects> {
        let asset_id = adjustment.asset_id().to_owned();
        let delta = adjustment.delta();

        let prev_total = self
            .totals
            .with(&asset_id, |total| *total)
            .unwrap_or(Volume::ZERO);

        let new_total = prev_total.checked_add(delta).map_err(|error| {
            Rejects::from(Reject::new(
                "BalanceLimitPolicy",
                RejectScope::Account,
                RejectCode::RiskLimitExceeded,
                "invalid adjustment total",
                error.to_string(),
            ))
        })?;

        if new_total > self.max_total {
            return Err(Rejects::from(Reject::new(
                "BalanceLimitPolicy",
                RejectScope::Account,
                RejectCode::RiskLimitExceeded,
                "cumulative adjustment exceeds limit",
                format!("asset {asset_id}: {new_total} > {}", self.max_total),
            )));
        }

        // Apply immediately so later adjustments in the same batch see the updated total.
        self.totals.with_mut(
            asset_id.clone(),
            || Volume::ZERO,
            |entry, _is_new| {
                *entry = new_total;
            },
        );

        // Register rollback: restore previous absolute value.
        // Safe because account adjustment batches are fully internal.
        let rollback_totals = Arc::clone(&self.totals);
        let rollback_asset = asset_id;

        mutations.push(Mutation::new(
            || {
                // Commit is empty: state was applied eagerly.
            },
            move || {
                // Rollback: restore absolute value captured before modification.
                rollback_totals.with_mut(
                    rollback_asset,
                    || Volume::ZERO,
                    |entry, _is_new| {
                        *entry = prev_total;
                    },
                );
            },
        ));

        Ok(openpit::pretrade::PolicyAccountAdjustmentResult {
            account_adjustments: Vec::new(),
            account_blocks: Vec::new(),
        })
    }
}

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Account Adjustments

Account Adjustments

Operation Types

Batch Semantics

Outcomes

Examples

Writing an Account Adjustment Policy with Rollback

Rollback Safety

Example: Balance Limit Policy

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally