Example Tutorials

What It Does

Establishes a FIX session with the paper trading server and demonstrates the event-driven connection lifecycle. After logon, queries cash balance + total balance via REST.

Why This Approach?

Key Concepts

• Event handlers first: Define on_logon(), on_logout(), on_reject() before client.connect().
• Handlers must accept **kw: payload schema can grow without notice.
• REST API access becomes available after logon (get_cash_balance, get_account_balance).
• Cleanup: uses os._exit(0) to avoid the known QuickFIX Python binding segfault on graceful exit.

Core Pattern

# 1. Define event handlers FIRST
def on_logon(session_id, **kw):
    print(f"✅ Connected: {session_id}")

# 2. Subscribe before connecting
client.on("fix:logon", on_logon)
client.on("fix:logout", on_logout)
client.on("fix:reject", on_reject)

# 3. Connect (non-blocking)
client.connect()

# 4. Wait for result
if client.wait_until_logged_on(timeout=10):
    cash  = client.get_cash_balance()
    total = client.get_account_balance()
else:
    print(f"Failed: {client.last_logon_error()}")

What It Does

Places multiple BUY / SELL orders sequentially with per-order acceptance gates, waits 15 seconds, then cancels only the orders still in active state (NEW / PENDING_NEW / PARTIALLY_FILLED). Rejected orders are tracked but skipped during cancel. Demonstrates the v0.2.6 tuple return from cancel_order — (is_terminal, status).

Why This Approach?

Key Concepts

• Tuple cancel return (v0.2.6+): is_terminal, status = client.cancel_order(cl_ord_id, timeout=5.0). is_terminal=False means the cancel didn't confirm within the timeout — may need manual retry.
• Rejected-order skipping: tracker marks rejected orders so the cancel loop iterates only the active set.
• Per-order events: compare cl_ord_id in handlers to filter relevant events.
• TIF default is DAY in v0.2.7 — Vietnam venues reject GTC. Override only on venues that support it.

Tuple-return cancel pattern

order_id = client.place_order(
    full_symbol=symbol, side=side, qty=qty, price=price, ord_type="LIMIT",
)
tracker.add_order(order_id, side, qty, price)

# Block until accepted OR rejected (handler sets one of these events)
tracker.accepted_events[order_id].wait(timeout=10)
if tracker.is_rejected(order_id):
    print("Skipping rejected order")
    return

# ...later, cancel only active ones:
is_terminal, final_status = client.cancel_order(order_id, timeout=5.0)
if is_terminal:
    print(f"Cancel confirmed: {final_status}")
else:
    print(f"Still pending: {final_status} (may need manual retry)")

Crash recovery (v0.2.4+, still in v0.2.7)

If the process crashes mid-flow, SQLite persistence (enabled by default via order_store_path="orders.db") lets you reload the active set on restart:

client.connect()
client.wait_until_logged_on(timeout=10)

pending = client.recover_pending_orders()
for p in pending:
    print(f"  {p['cl_ord_id']}: {p['side']} "
          f"{p['qty']}x {p['symbol']} @ {p['price']}")
    client.cancel_order(p["cl_ord_id"])      # tuple return

What It Does

D1 (main) sells contracts; D2 (counterparty) buys to match unconditionally. Two scenarios:

1. Full fill: D1 sells 5 contracts, D2 buys 5 in one shot → complete match.
2. Partial fills: D1 sells 5, D2 buys 1+1+1+2 → four partial-fill events tracked end-to-end.

Why This Approach?

Key Concepts

• Sub-account switching: client.use_sub_account("D2") context manager — thread-safe, scoped to the with block.
• Per-order tracker map: route fill events to per-order state by cl_ord_id.
• Partial-fill vs terminal-fill: compare cum_qty with the original qty.
• Both fix:order:partial_fill and fix:order:filled carry the same payload; bind one handler to both if you don't care about terminality.

What It Does

Uses RedisMarketDataClient.query(instrument) for synchronous-style quote lookups — useful for batch snapshots, settling decisions, or one-off reads. No subscription, no pub/sub overhead.

Pattern

from paperbroker.market_data import RedisMarketDataClient

client = RedisMarketDataClient(
    host=os.getenv("MARKET_REDIS_HOST"),
    port=int(os.getenv("MARKET_REDIS_PORT", 6379)),
    password=os.getenv("MARKET_REDIS_PASSWORD"),
    merge_updates=True,
)

quote = await client.query("HNXDS:VN30F2606")
if quote:
    print(f"latest={quote.latest_matched_price}, "
          f"bid={quote.bid_price_1}, ask={quote.ask_price_1}")

What It Does

Subscribes to a Redis pub/sub channel for real-time quote updates. Default MERGED mode delivers full snapshots per tick — unchanged fields carry forward; the caller doesn't reconstruct state. Flip merge_updates=False for RAW mode (deltas only; lower bandwidth, more caller work).

Key Concepts

• The callback fires from the client's consumer task — keep it fast; offload heavy work to a ThreadPoolExecutor.
• Channel format: {exchange}:{symbol}.
• For Kafka transport (recommended for multi-instrument alphas), see Example 08.

Pattern

def on_quote(instrument, quote):
    print(f"{instrument}: bid={quote.bid_price_1}, ask={quote.ask_price_1}")

await client.subscribe("HNXDS:VN30F2606", on_quote)
# Stream runs until close()
await client.close()

What It Does

Subscribes to a Kafka topic for real-time quotes. Topic format: {env_id}.{exchange}.{symbol} (e.g. real.HNXDS.VN30F2606). Authentication via SASL/PLAIN. Kafka requires await client.start() AFTER subscribe() to start the background consumer; Redis no-ops it.

Key Concepts

• Required env vars: PAPERBROKER_KAFKA_BOOTSTRAP_SERVERS, PAPERBROKER_KAFKA_USERNAME, PAPERBROKER_KAFKA_PASSWORD, PAPERBROKER_ENV_ID.
• Higher throughput than Redis; reliable on back-month contracts where Redis pub/sub is sparse.
• For an alpha driven by Kafka MD, see Example 12 (pair-spread).

Pattern

from paperbroker.market_data import KafkaMarketDataClient

client = KafkaMarketDataClient(
    bootstrap_servers=os.getenv("PAPERBROKER_KAFKA_BOOTSTRAP_SERVERS"),
    username=os.getenv("PAPERBROKER_KAFKA_USERNAME"),
    password=os.getenv("PAPERBROKER_KAFKA_PASSWORD"),
    env_id=os.getenv("PAPERBROKER_ENV_ID", "real"),
    merge_updates=True,
)
await client.subscribe("HNXDS:VN30F2606", on_quote)
await client.start()   # required AFTER subscribe (Kafka only)
# ... later:
await client.stop()

What It Does

Demonstrates the three REST endpoints that surface account history:

• get_portfolio_by_sub() — current positions + unrealized PnL.
• get_orders(start, end) — order history (intents).
• get_transactions_by_date(start, end) — executed fills (realized).

Conceptual split

Pattern

portfolio = client.get_portfolio_by_sub("main")
for pos in portfolio.get("items", []):
    print(f"  {pos['instrument']}: {pos['quantity']} @ {pos['currentPrice']}")

start = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
end   = datetime.now().strftime("%Y-%m-%d")

orders = client.get_orders(start, end)
txs    = client.get_transactions_by_date(start, end)
print(f"Orders: {len(orders.get('items', []))} | Txs: {len(txs)}")

What It Does

Calls get_max_placeable(symbol, price, side) to compute the maximum order quantity given current cash, margin requirements, fees, and existing positions. Useful for risk-checks before any place_order — avoids rejected orders due to over-sizing.

Key Concepts

• BUY needs cash; SELL might be unlimited (if you already hold) or capped (short selling rules).
• Returns {maxQty, perUnitCost, remainCash, unlimited}; unlimited=True ⇒ no cash cap (rare).
• Industry-standard safety margin: place at most 80% of maxQty to absorb price drift and fee variance.

Pattern

result = client.get_max_placeable(symbol, price=1200.0, side="BUY")
max_qty = result["maxQty"]
if result["unlimited"]:
    print("No limit")
else:
    print(f"Max BUY: {max_qty} contracts @ {result['perUnitCost']} each")
    safe_qty = int(max_qty * 0.8)            # 80% safety margin
    client.place_order(symbol, "BUY", safe_qty, price=1200.0)

What It Does

RSI(14) on 1-minute bars. Entry: BUY when RSI < oversold (30), SELL when > overbought (70). Exit: close long when RSI crosses back above 50; close short below 50. Pure-Python RSI — no numpy / pandas required.

Key Concepts

• 4 hooks: get_indicators, get_signals, get_entry_price, get_quantity. The framework wires bar aggregation, fill tracking, and cleanup.
• Exit logic lives in get_signals — framework has no built-in TP/SL. Inspect ctx.positions, emit opposite-side Signal when exit fires.
• declared_params set typo-guards param keys at __init__.
• Run via YAML: paperbroker run --config examples/09_alpha_rsi_1m.yaml (requires pip install 'paperbroker_client[cli]').

Hook chain

class RSI1MAlpha(SignalDrivenAlpha):
    declared_params = {"rsi_period", "oversold", "overbought"}

    def get_indicators(self, ctx):
        sym = self.config.instruments[0]
        closes = [b.close for b in ctx.bars[sym]]
        return {"rsi": _rsi(closes, int(self.config.params["rsi_period"]))}

    def get_signals(self, indicators, ctx):
        sym, rsi = self.config.instruments[0], indicators["rsi"]
        if rsi is None: return []
        pos = ctx.positions.get(sym)
        if pos and abs(pos.quantity) > 1e-9:                        # exit
            if pos.quantity > 0 and rsi > 50: return [Signal(sym, "SELL", tag="exit_long")]
            if pos.quantity < 0 and rsi < 50: return [Signal(sym, "BUY",  tag="exit_short")]
            return []
        if rsi < 30: return [Signal(sym, "BUY",  tag="entry")]      # entry
        if rsi > 70: return [Signal(sym, "SELL", tag="entry")]
        return []

    def get_entry_price(self, signal, ctx): return ctx.bars[signal.symbol][-1].close
    def get_quantity(self, signal, ctx):    return self.config.qty_for(signal.symbol)

What It Does

BUY on lower-band touch, SELL on upper-band touch (SMA ± kσ). Exit when price returns to mid band. StateStore sticky-lock: remembers the last fired signal in JSON so a restart doesn't re-enter on the same band touch.

Key Concepts

• state_path in AlphaConfig enables self.state_store (JSON K/V with atomic writes).
• Accessing state_store when state_path=None raises RuntimeError — defensive code can check self._state_store is not None.
• Same hook signatures as the RSI example — just with the StateStore wired in.

What It Does

Pedagogical replacement for the removed OCO infrastructure. Trivial SMA-crossover entry, plus 4 exit conditions combined:

1. Take-profit at fixed +5 points → TakeProfitSignal
2. Stop-loss at fixed −3 points → StopLossSignal
3. Peak-to-trough drawdown → CloseSignal(tag="drawdown")
4. Time-of-day flatten at 14:44:30 Asia/Ho_Chi_Minh → CloseSignal(tag="eod")

Why subtypes

Best practice

Check exits BEFORE entries inside get_signals — otherwise a position that meets both conditions flips on every bar.

What It Does

Trades the spread between two correlated derivatives (e.g. VN30F1M vs VN30F2M). When the rolling-window z-score of close(A) − close(B) is extreme, enters both legs atomically (one BUY + one SELL). Exits when the z-score reverts toward zero.

Key Concepts

• Joint trigger: framework debounces bar arrivals across both symbols (bar_window_ms=500); one AlphaContext is built per burst, get_signals sees both legs together.
• get_signals returns a list of Signals — multi-leg atomic dispatch is the pattern v0.2.7 was designed around.
• Per-symbol qty: qty={f1m: 1, f2m: 1}.
• Use Kafka MD — Redis pub/sub is too sparse on back months for joint trigger to fire reliably.

Atomic 2-leg dispatch

def get_signals(self, indicators, ctx):
    z = indicators["z"]
    a, b = self.config.instruments
    if z is None: return []

    # Entry: spread too high -> short A, long B (atomic)
    if z > self.config.params["entry_z"]:
        return [Signal(a, "SELL", tag="entry_short_spread"),
                Signal(b, "BUY",  tag="entry_short_spread")]
    if z < -self.config.params["entry_z"]:
        return [Signal(a, "BUY",  tag="entry_long_spread"),
                Signal(b, "SELL", tag="entry_long_spread")]
    return []

alpha = SpreadPairAlpha.from_paper(
    instruments=[f1m, f2m],
    sub_account="main", timeframe="1m",
    qty={f1m: 1, f2m: 1},
    params={"window": 20, "entry_z": 2.0, "exit_z": 0.5},
)

What It Does

Short-term momentum on the VN30 index — but executes on VN30F1M (the futures contract). The index isn't tradeable; the future is. Demonstrates the Signal vs OrderRequest split: Signal.symbol is the idea source; the order can target a different symbol via the plan_orders override.

plan_orders override

def plan_orders(self, signal, ctx):
    # Signal.symbol = index; order trades the future
    exec_sym = self.config.params["execution_symbol"]
    bars = ctx.bars[exec_sym]
    if not bars: return []
    return [OrderRequest(
        symbol=exec_sym,
        side=signal.side,
        qty=self.config.qty_for(exec_sym),
        price=bars[-1].close,
        ord_type="LIMIT",
        parent_signal_id=None,    # framework assigns
    )]

What It Does

SMA-crossover entry; plan_orders splits the total quantity into slice_count legs at price levels [bar.close − i × step] (or + for SELL). Each slice carries the same parent_signal_id (auto-assigned) so the framework attributes fills to one logical entry.

Key Concepts

• plan_orders returns N OrderRequests — framework places each, tracking lifecycle individually.
• ctx.signals[parent_id] aggregates fill state across all legs (total_filled_qty, remaining_qty, VWAP avg_fill_px).
• Multi-leg auto-OCO is disabled by design — aggregate exit semantics live inside get_signals.

What It Does

Quote both sides of the book around mid (mid ± spread_pts). Skip a side when inventory cap hit. Cancel-then-place pattern: when mid drifts > price_drift_ticks, cancel the working quote on that side; the next evaluation re-quotes.

Quote-driven pattern

• trigger_on_bar = False + trigger_on_quote = True — hook chain fires on quote ticks, not bar close.
• should_evaluate_on_quote throttles to actual price changes — skip ticks where the matched price didn't move.
• Framework no longer blocks per-symbol (v0.2.7 Milestone E removed the implicit single-entry gate) — alpha emits both bid & ask quotes as [Signal(BUY, "bid"), Signal(SELL, "ask")] on every refresh.
• ctx.open_orders[sym] is a list of OrderState — inspect .side and .is_working to decide whether to cancel-then-place.

Backend

MM is quote-driven and needs a reliable per-tick stream. Default Redis pub/sub is too sparse on back-months; use Kafka MD for production-like MM (see Example 08 + market-data backend guidance).