Skip to content
Connect > Client Libraries > Use Cases

Use case: a realtime risk and P&L loop

Goal. A program watches trades as they happen, recomputes a running risk or P&L figure per update, and writes the result back into a Center table, where risk managers, other services, or a dashboard pick it up live. This is the closed loop: subscribe, compute, write.

The shape

1
trades  →  subscribe (Center)  →  compute  →  SQL INSERT/UPDATE (worker thread)  →  RISK table  →  others subscribe

One client does both halves: a Center config to receive trades and a JDBC config to write the recomputed rows.

Write results with SQL from a worker thread, not publish, and not from the callback

Two rules collide here, and the resolution is the whole point of this use case:

  • publish does not populate a Center table (see the market-data use case), so the result must be written with SQL INSERT/UPDATE.
  • Do not run the SQL write on the callback thread. The callback runs on the client's I/O (delivery) thread; a slow or same-connection synchronous call there stalls update delivery (and a same-connection call can block). Hand the write to your own thread.

The working pattern: in on_update do the fast arithmetic, then hand the row to your own worker thread, which runs the sql write. This was confirmed on a live instance.

1. Connect for both directions

1
2
3
4
5
6
7
import pyami.rt.ami_client as ac

node = ac.AmiCluster("N1", "localhost")
node.center_config = ac.AmiSubscriptionConfig("3270", True, "demo", "demo123")  # receive trades
node.jdbc_config   = ac.AmiDatabaseConfig("3280", True, "demo", "demo123", 60000, 10000)  # write results
# point .shards[0].ssl at your certs on a secured server (see Configuration)
client = ac.AmiClient()

1
2
3
4
Cluster node("N1", "localhost");
node.subscription_config() = SubscriptionConfig("3270", true, "demo", "demo123");
node.database_config()     = DatabaseConfig("3280", true, {}, "demo", "demo123");
Client client;

2. Recompute on each trade, write the result off-thread

Keep a running position per symbol. on_update updates it in memory (fast, on the callback thread) and queues the recomputed row; a worker thread drains the queue and writes each row with sql.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import queue, threading

class RiskEngine:
    def __init__(self, client):
        self.client = client
        self.pos = {}                       # symbol -> qty
        self.q = queue.Queue()
        threading.Thread(target=self._writer, daemon=True).start()

    def on_update(self, client, name, update):     # I/O thread, keep it fast, no sql() here
        if update.action == 3:                      # a deleted trade, ignore in this example
            return
        t = dict(zip(update.field_names, update.row_data))
        sym = t["symbol"]
        self.pos[sym] = self.pos.get(sym, 0) + t["quantity"]
        self.q.put((sym, self.pos[sym], self._pnl(sym)))   # hand off, never call sql() here

    def _writer(self):
        while True:
            sym, position, pnl = self.q.get()
            self.client.sql("N1",
                f'INSERT INTO RISK (symbol, position, pnl) '
                f'VALUES ("{sym}", {position}, {pnl})')

engine = RiskEngine(client)
client.add_client_listener(engine)
client.add_cluster(node, sync=True)
client.subscribe("N1", ["TRADES"])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
class RiskEngine : public ClientListener {
    Client & client_;
    std::unordered_map<std::string, std::int64_t> pos_;
    // a thread-safe queue drained by a worker thread that calls client_.sql(...)
public:
    explicit RiskEngine(Client & c) : client_(c) { /* start worker thread */ }
    void on_update(Client &, std::string_view, const CenterIncrementalUpdate & u) override {
        if (u.action_ == 3) return;
        std::string sym = field<std::string>(u, "symbol");   // copy out of the callback buffers
        pos_[sym] += field<std::int64_t>(u, "quantity");
        enqueue(sym, pos_[sym], pnl(sym));                   // worker thread runs the sql INSERT
    }
};

If the RISK table has a primary-key index on symbol, re-inserting a symbol upserts (updates the row in place); without one, use UPDATE ... WHERE symbol == "..." for existing symbols. Copy the update's fields before handing them to the worker, they are valid only during the callback.

3. Anyone can watch the result

Another program, a risk dashboard, an alerting service, subscribes to RISK the same way the market-data dashboard does (merging on id), and sees each recomputed figure as it lands.

Optional: a command to trigger a recompute

Register a named command so the platform (or a user in the UI) can ask your engine to recompute or flatten a book on demand: handle it in on_command and reply with send_command_response. This is how "kill switch" or "recalculate" buttons reach a native engine.

What this shows

  • The loop closes in one client: subscribe on the Center connection, write on the JDBC connection.
  • Your firm's logic stays in your code, in the language your risk model already uses; the platform carries the data in and the results out.
  • The threading model is the crux: compute on the callback thread, and hand the write to a worker thread so a slow or same-connection SQL call never stalls update delivery.