CPZ Python SDK

Strategy Framework, Backtesting Engine, Risk Guards, Execution Algorithms, and Multi-Broker Trading

What's New in v4.7.0 — Institutional Trust Suite

Five coordinated controls land together, each aligned to SEC Rule 15c3-5 (pre-trade) and SEC 17a-4 / FINRA 4511 (records) principles. This is a controls framework, not a claim of regulatory certification.

• Bounded Autonomy (client.mandates) — trading mandates that bound order flow per scope: symbol universe, per-order notional and quantity caps, gross exposure, leverage, daily-loss limit, and a mandatory expiry for live scopes. Enforced at every first-party order and deploy surface, backed by a hash-chained local audit ledger (cpz.audit) and mandate / killswitch / audit CLI subcommands.
• Backtest Trust Layer (client.backtests) — every backtest can emit a run card and validation artifact: metrics, benchmark context, data provenance, a deterministic config hash, and point-in-time / OHLC-sanity / lookahead-bias checks. The "SDK-attested" verdict is recomputed server-side, so a client can never self-certify.
• Factor Library (client.factors) — fetch the platform factor catalog and bench any factor over a universe and period (IC, rank IC, IR, turnover, with alive / reversed / dead status). Formulas run through an AST-whitelisted evaluator, never eval.
• Strategy Health (client.health) — a falsifiable-hypothesis registry plus a decay lifecycle (active / monitoring / decayed / disabled) computed from rolling Sharpe and IC. It records only; it never stops or modifies a deployment.
• Shadow Account (client.shadow) — parse broker execution history (Alpaca, IBKR Flex, generic CSV) into normalized fills, compute behavioral diagnostics (FIFO round trips, win rate, holding periods, disposition effect), and replay an honestly-bounded counterfactual.

Behavioral changes — read before upgrading:

• Live orders now require CPZ platform connectivity. The kill switch and mandate checks fail closed: if the control plane is unreachable, a live order raises rather than proceeding unchecked. Paper orders warn and proceed. Risk-reducing closes are exempt by design.
• validate=True is the new backtest default. Data-quality checks and the lookahead gate now run unless you pass validate=False.
• Removed the unauthenticated hardcoded-IP engine REST fallback, which cannot coexist with mandate enforcement.

Recent releases

• v4.6.6 — Bloomberg Data License market-data provider (client.data.bloomberg); order-fill durability.
• v4.6.3 — real, vendor-agnostic options chain with greeks (options() / get_option_chain()), routed through the CPZ backend instead of hitting Alpaca directly.
• v4.6.0 — implied-volatility history and IV rank (client.data.get_iv_history / iv_rank), vendor-agnostic and capability-based.
• v4.2.0 — client.data.scout(source, **params): one-call execution of any Data Scout starter snippet, credentials resolved server-side from your stored API connections.
• v4.1.0 — institutional FIX 4.4 client (cpz.fix) for direct broker connectivity, sync and async.
• v4.25.0: research Compute Fabric client and truthful local QAOA simulation; earlier real-QPU claims corrected.
• v3.0.0 — Strategy Framework, Backtest Engine, Pre-Trade Risk Guard, TWAP/VWAP/Iceberg execution algos.

See CHANGELOG.md for the full reconciled history (v2.4.0 → v4.7.0).

Overview

The CPZ Python SDK is the unified interface for the CPZ quantitative trading platform. Write strategies once and run them in backtest or live mode with zero code changes.

Installation

pip install cpz-ai                    # core SDK (includes classical quantum-inspired optimizers)
pip install cpz-ai[risk]              # + statsmodels for advanced risk analytics
pip install cpz-ai[quantum]           # + dwave-neal for simulated annealing
pip install cpz-ai[quantum-braket]    # + Amazon Braket local QAOA simulation (CPU)
pip install cpz-ai[all]               # + all optional dependencies

Quantum Portfolio Optimization

QUBO formulations support classical exact and annealing solvers, plus an experimental Braket local CPU simulator. Remote managed simulators and physical QPUs are disabled in cpz-quant 1.1.0. Earlier hardware and cost-tracking claims were unsupported by the implementation and are superseded here.

from cpz.portfolio import BraketSolver, qubo_portfolio_selection

# No AWS credentials or provider charges. Local CPU resources still have cost.
solver = BraketSolver(device="local", shots=256, max_iter=8, seed=7)
result = qubo_portfolio_selection(returns, target_k=2, solver=solver)

This does not demonstrate quantum advantage. QHRP permutation problems have N squared binary variables; local QAOA is bounded to 20 variables.

Compute Fabric research API (v4.25.0)

client.compute provides advisory recommendations, explicit small CPU hedge solves, batch planning, and benchmark uploads. client.quantum is an alias for the research interface, not physical QPU execution.

# Existing authenticated CPZClient. All numbers below are synthetic.
proposal = client.compute.optimize_hedge(
    backend="edge_wasm", algorithm="brute_force", request_id="hedge-demo-1",
    book_greeks={"DEMO": {"delta": 1, "gamma": 0, "vega": 0}},
    candidates=[{
        "symbol": "DEMO-HEDGE", "underlying": "DEMO", "lot": 1,
        "greeks_per_lot": {"delta": -1, "gamma": 0, "vega": 0},
        "notional_per_lot": 100, "spread_bps": 1,
    }],
    cardinality=1, penalty=10,
    greek_weights={"delta": 1, "gamma": 1, "vega": 1},
)
assert proposal.production_eligible is False

The endpoint requires platform schema activation, entitlement and a spending policy. It fails closed when these are unavailable. Keep the same request ID after an uncertain response; use a new ID only for intentional new work. Run uploads are caller-reported evidence, never billing authorization. Reported provider costs are not verified invoices.

Proposals validate linear Greeks only and cannot emit orders. Live orders declaring a research compute_run_id are blocked even in shadow mode. Untagged user-created orders use existing controls; manually copied numbers cannot be traced. GPU execution and physical-QPU dispatch remain unavailable.

Strategy Framework

Write your strategy once as a Strategy subclass. The same code runs against the backtest engine or live broker with zero changes.

Define a Strategy

from cpz import Strategy, StrategyConfig, StrategyRunner, CPZClient

class MomentumConfig(StrategyConfig):
    fast_period: int = 10
    slow_period: int = 30
    trade_size: float = 100

class MomentumStrategy(Strategy):
    def on_start(self):
        self.subscribe_bars(self.config.instruments[0], "1D")

    def on_bar(self, bar):
        prices = self.cache.bars(bar.symbol)
        if len(prices) < self.config.slow_period:
            return
        fast_ma = sum(b.close for b in prices[-self.config.fast_period:]) / self.config.fast_period
        slow_ma = sum(b.close for b in prices[-self.config.slow_period:]) / self.config.slow_period

        if fast_ma > slow_ma and not self.portfolio.is_long(bar.symbol):
            self.buy(bar.symbol, qty=self.config.trade_size)
        elif fast_ma < slow_ma and self.portfolio.is_long(bar.symbol):
            self.flatten(bar.symbol)

    def on_order_filled(self, event):
        self.log.info(f"Filled {event.symbol} @ {event.fill_price}")

    def on_position_closed(self, event):
        self.log.info(f"Closed {event.symbol}, realized P&L: ${event.realized_pnl:,.2f}")

Backtest

client = CPZClient()
runner = StrategyRunner(client, mode="backtest")
runner.add_strategy(MomentumStrategy(MomentumConfig(instruments=["AAPL"])))

result = runner.backtest(
    start="2024-01-01",
    end="2024-12-31",
    initial_cash=100_000,
    commission_pct=0.0003,    # 3 bps
    slippage_pct=0.0001,      # 1 bp
)
print(result.summary())

Output:

============================================================
  BACKTEST RESULTS
============================================================
  Period:            2024-01-01 → 2024-12-31
  Initial Cash:      $100,000.00
  Final Equity:      $112,450.00
  Total Return:      +12.45%
------------------------------------------------------------
  Sharpe Ratio:      1.842
  Sortino Ratio:     2.315
  Max Drawdown:      -4.23%
------------------------------------------------------------
  Total Trades:      47
  Win Rate:          63.8%
  Profit Factor:     2.14
============================================================

Live Trading (Same Code!)

runner = StrategyRunner(client, mode="live")
runner.add_strategy(MomentumStrategy(MomentumConfig(instruments=["AAPL"])))
runner.run()  # Blocks until Ctrl+C

Strategy Lifecycle Hooks

Strategy Actions

self.buy("AAPL", qty=100)                    # Market buy
self.sell("AAPL", qty=50)                     # Market sell
self.buy("AAPL", qty=10, order_type="limit", limit_price=150.00)
self.flatten("AAPL")                          # Close position
self.flatten_all()                            # Close all positions
self.cancel_all_orders()                      # Cancel open orders
self.emit_signal("crossover", value=1.0)      # Emit custom signal
self.subscribe_bars("AAPL", "1Min")           # Subscribe to bars
self.subscribe_quotes("AAPL")                 # Subscribe to quotes

Strategy Context

Every strategy has access to:

self.cache.bars("AAPL")           # Cached bar history
self.cache.latest_bar("AAPL")     # Most recent bar
self.portfolio.equity()           # Current equity
self.portfolio.cash               # Available cash
self.portfolio.is_long("AAPL")    # Position check
self.portfolio.positions()        # All positions
self.clock.utc_now()              # Current time (real or simulated)
self.log.info("message")          # Structured logging
self.config.instruments           # Strategy config

Pre-Trade Risk Guard

Intercepts every order before it reaches the broker. Any violation raises RiskGuardViolation and the order is rejected.

from cpz import CPZClient

client = CPZClient()

# Attach risk guard to execution
client.execution.set_risk_guard(
    max_order_value=50_000,         # Reject orders > $50K
    max_position_pct=0.10,          # Max 10% of equity per position
    max_daily_loss=-5_000,          # Halt trading if daily P&L < -$5K
    max_open_orders=20,             # Max concurrent open orders
    blocked_symbols=["GME", "AMC"], # Blacklist
    cooldown_seconds=5.0,           # Min 5s between orders per symbol
    max_orders_per_minute=30,       # Rate limit
)

# Orders are now validated before submission
order = client.execution.order(symbol="AAPL", qty=10, side="buy", strategy_id="my-strat")

Or use directly in a strategy:

from cpz import RiskGuard

class SafeStrategy(Strategy):
    def on_start(self):
        guard = RiskGuard(max_order_value=25_000, max_daily_loss=-2_000)
        self._router.set_risk_guard(guard)
        self.subscribe_bars("AAPL", "1D")

Execution Algorithms

Split large orders into smaller slices for better execution.

from cpz.execution.algos import TWAPAlgorithm, VWAPAlgorithm, IcebergAlgorithm

# TWAP — equal slices over time
algo = TWAPAlgorithm()
slices = algo.generate_slices(order_request, {
    "duration_minutes": 30,
    "num_slices": 10,
})

# VWAP — volume-weighted slicing with intraday profile
algo = VWAPAlgorithm()
slices = algo.generate_slices(order_request, {
    "duration_minutes": 60,
    "num_slices": 13,
})

# Iceberg — hidden quantity
algo = IcebergAlgorithm()
slices = algo.generate_slices(order_request, {
    "display_qty": 100,  # Only show 100 shares at a time
})

Typed Domain Model

Precision-safe value types that reject NaN, Infinity, and invalid data at construction time.

from cpz import Price, Quantity

# Validated, immutable
price = Price("150.25")                # From string (exact)
price = Price.from_float(150.25, precision=2)  # From float
qty = Quantity.from_int(100)

# Type-safe arithmetic
total = price * qty                    # Returns Money("15025.00", "USD")
spread = Price("150.50") - Price("150.25")  # Returns Price("0.25")

# Fail-fast on invalid data
Price(float("nan"))    # ValueError: Price must be finite
Quantity(-1)           # ValueError: Quantity cannot be negative

Event Bus

Thread-safe pub/sub with wildcard topic matching.

from cpz import EventBus

bus = EventBus()

# Subscribe with wildcards
bus.subscribe("bar.AAPL.*", handle_aapl_bars)     # Any timeframe for AAPL
bus.subscribe("order.**", handle_all_orders)       # All order events

# Publish
bus.publish("bar.AAPL.1Min", bar_event)

Quick Start

from cpz import CPZClient

client = CPZClient()

# ── Risk Analytics ──────────────────────────────────────
sharpe = client.risk.sharpe(daily_returns)
snapshot = client.risk.compute(daily_returns, spy_returns, weights)
mc = client.risk.monte_carlo(daily_returns, num_simulations=10000)

# ── Trading ─────────────────────────────────────────────
client.execution.use_broker("alpaca", environment="paper")
order = client.execution.order(symbol="AAPL", qty=10, side="buy", strategy_id="my-strat")

# ── Market Data ─────────────────────────────────────────
bars = client.data.bars("AAPL", timeframe="1D", limit=100)
quotes = client.data.quotes(["AAPL", "MSFT", "GOOGL"])
gdp = client.data.economic("GDP")

# ── Simons (Quant Strategist) ───────────────────────────
response = client.simons.chat("Analyze AAPL for momentum trading")
print(response.content)

Risk Analytics

Comprehensive quantitative risk computation powered by numpy/scipy. Install with pip install cpz-ai[risk].

The risk-free rate is automatically fetched from FRED (3-Month Treasury) on first use. All methods accept daily returns as a list of floats.

Core Metrics

from cpz import CPZClient

client = CPZClient()

# Individual metrics
sharpe = client.risk.sharpe(daily_returns)               # Annualized Sharpe ratio
sortino = client.risk.sortino(daily_returns)              # Sortino (downside vol only)
vol = client.risk.volatility(daily_returns)               # Annualized volatility (%)
mdd = client.risk.max_drawdown(daily_returns)             # Maximum drawdown (%)
b = client.risk.beta(daily_returns, spy_returns)          # Beta vs benchmark
a = client.risk.alpha(daily_returns, spy_returns)         # Jensen's alpha (annualized)

# Full risk snapshot (all metrics at once)
snapshot = client.risk.compute(
    daily_returns=portfolio_returns,
    benchmark_returns=spy_returns,
    position_weights={"AAPL": 0.4, "NVDA": 0.35, "BTC/USD": 0.25},
    total_exposure=100000,
)

Sharpe Inference (Lopez de Prado Framework)

Statistical significance testing for the Sharpe ratio. Implements all 5 corrections from Lopez de Prado (2012, 2018):

1. Non-normality — SE adjusted for skewness and kurtosis
2. Serial correlation — Lo (2002) autocorrelation adjustment
3. Probabilistic Sharpe Ratio (PSR) — P(true Sharpe > benchmark)
4. Deflated Sharpe Ratio (DSR) — Multiple-testing correction
5. Minimum Track Record Length — How long before the Sharpe is credible?

inference = client.risk.sharpe_inference(daily_returns, num_trials=20)
print(f"PSR (prob true SR > 0): {inference.psr:.1%}")
print(f"DSR (adjusted for 20 trials): {inference.deflated_sharpe}")
print(f"Min track record needed: {inference.min_track_record_months} months")

Value at Risk

var_95 = client.risk.parametric_var(daily_returns, confidence=0.95)
hist_var = client.risk.historical_var(daily_returns, confidence=0.95)
mc = client.risk.monte_carlo(daily_returns, num_simulations=10000, horizon_days=5)

Advanced Analytics

dd = client.risk.drawdown_analysis(daily_returns)        # Drawdown decomposition
tail = client.risk.tail_risk(daily_returns)              # Skewness, kurtosis, Cornish-Fisher VaR
sizing = client.risk.position_size(100000, returns)      # Kelly criterion + vol targeting
exposure = client.risk.factor_exposure(returns, factors) # OLS factor decomposition
impacts = client.risk.stress_test_all(weights)           # 7 historical crisis scenarios
rolling = client.risk.rolling_metrics(returns, spy, 20)  # Rolling Sharpe, vol, beta

Trading

Broker Configuration

from cpz import CPZClient

client = CPZClient()

client.execution.use_broker("alpaca", environment="paper")
client.execution.use_broker("alpaca", environment="live")
client.execution.use_broker("ibkr", environment="paper")

Order Placement

# Simple order
order = client.execution.order(
    symbol="AAPL",
    qty=10,
    side="buy",
    strategy_id="my-strategy"
)

# Full control
from cpz import OrderSubmitRequest, OrderSide, OrderType, TimeInForce

request = OrderSubmitRequest(
    symbol="AAPL",
    side=OrderSide.BUY,
    qty=10,
    order_type=OrderType.LIMIT,
    time_in_force=TimeInForce.GTC,
    limit_price=150.00,
    strategy_id="my-strategy"
)
order = client.execution.submit_order(request)

Account and Positions

account = client.execution.get_account()
print(f"Buying Power: ${account.buying_power:,.2f}")

positions = client.execution.get_positions()
for pos in positions:
    print(f"{pos.symbol}: {pos.qty} shares @ ${pos.avg_entry_price}")

Execution Gateway Routing

With CPZ_PROXY_ENABLED=1 in the environment (hosted strategy runs set it), use_broker("alpaca") and the implicit Alpaca paper default behind client.execution.order(...) route through the CPZAI execution gateway (cpz/orders/execute) instead of talking to Alpaca from your process: broker credentials stay server-side, the pre-trade guard runs on every submission, and each order carries a client_order_id idempotency key. Passing api_key_id, api_secret_key or oauth_token to use_broker keeps the native adapter. router.routed_via tells you which path is active.

import os

os.environ["CPZ_PROXY_ENABLED"] = "1"
client.execution.use_broker("alpaca", environment="paper")
print(client.execution.router.routed_via)  # "cpz_proxy"

Every OrderSubmitRequest gets client_order_id = "cpz-<strategy8>-<uuid12>" unless you supply one (at most 64 characters from [A-Za-z0-9_.:-]). The returned Order carries it, plus idempotent_replay=True when the gateway answered from an earlier submission of the same id instead of placing a second broker order. Gateway outcomes are typed:

from cpz.execution.cpz_proxy import (
    CPZBrokerRejected,
    CPZOrderLaneBusy,
    CPZSubmitUnconfirmed,
)

try:
    order = client.execution.order(symbol="AAPL", qty=10, side="buy", strategy_id="my-strategy")
except CPZSubmitUnconfirmed as exc:
    # The order may be live under exc.client_order_id. Look it up with
    # list_orders() or resend the identical request; never use a fresh id.
    ...
except CPZOrderLaneBusy as exc:
    # Another submission holds this symbol's lane and the SDK already retried
    # once. Retry the identical request after exc.retry_after_ms.
    ...
except CPZBrokerRejected as exc:
    # Definitive broker rejection: exc.code (for example wash_trade_rejected)
    # and exc.broker_code carry the reason. Not retryable as-is.
    ...

A timeout or connection error mid-submit is not treated as "not placed": the SDK looks the order up by client_order_id (twice, 500 ms apart) and returns it if the broker id is bound, and raises CPZSubmitUnconfirmed otherwise.

Quantitative Libraries (v2.5.0+)

Local Indicators

25+ indicators computed locally (no API calls):

from cpz.indicators import sma, ema, rsi, macd, bollinger, atr, vwap

fast = ema(prices, period=10)
slow = ema(prices, period=30)
rsi_val = rsi(prices, period=14)
bb = bollinger(prices, period=20)   # .upper, .middle, .lower
macd_val = macd(prices)             # .macd, .signal, .histogram

Portfolio Optimization

12 optimizers including HRP, Black-Litterman, and Mean-CVaR:

from cpz.portfolio import mean_variance, risk_parity, hierarchical_risk_parity

weights = risk_parity(returns_df)
weights = hierarchical_risk_parity(returns_df)

Signal Construction

from cpz.signals import vol_target, kelly, regime_filter, max_positions

sized = vol_target(raw_signals, returns, target_vol=0.15)
filtered = regime_filter(signals, returns, window=60)
limited = max_positions(signals, max_n=10)

Alpha Research

from cpz.alpha import information_coefficient, signal_decay, quantile_returns

ic = information_coefficient(signals, forward_returns)
decay = signal_decay(signals, returns, max_lag=20)

HFT Engine

Deploy strategies to the Rust HFT engine for autonomous microsecond-latency execution.

status = client.engine.status()
client.engine.deploy(
    strategy_id="my-strategy",
    symbols=["AAPL", "NVDA"],
    broker="alpaca",
    environment="paper",
)
client.engine.stop("my-strategy")

Data

# Stock/crypto bars
bars = client.data.bars("AAPL", timeframe="1D", limit=100)

# Multi-symbol history for backtesting
df = client.data.history(["AAPL", "MSFT", "NVDA"], timeframe="1D", limit=252)

# Quotes, news, options, economic data, filings, sentiment
quotes = client.data.quotes(["AAPL", "MSFT"])
news = client.data.news("TSLA", limit=10)
gdp = client.data.economic("GDP")
filings = client.data.filings("AAPL", form="10-K")
sentiment = client.data.sentiment("GME")

# Massive (formerly Polygon): stocks, options, indices, futures, crypto, forex
bars  = client.data.massive.get_bars("AAPL", timeframe="1d")
chain = client.data.massive.get_options_chain("AAPL")
quote = client.data.massive.realtime.get_quote("AAPL")   # refuses DELAYED data

# 18 data providers: Massive, Alpaca, TwelveData, FRED, SEC EDGAR,
# Yahoo Finance, Databento, CoinGecko, Finnhub, Alpha Vantage, and more

Data Connections Guide — how to connect each provider, then fetch, pull and stream from it (key resolution, entitlements, streaming support, troubleshooting).

Simons — Quantitative Trading Strategist

response = client.simons.chat("Analyze AAPL for momentum trading")
print(response.content)

for chunk in client.simons.stream("Write a mean-reversion backtest"):
    if chunk.type == "text":
        print(chunk.content, end="", flush=True)

Architecture

CPZClient
├── Strategy Framework (NEW in v3.0.0)
│   ├── Strategy          Base class with 20+ lifecycle hooks
│   ├── StrategyConfig    Serializable configuration
│   ├── StrategyRunner    Live and backtest orchestration
│   ├── BacktestEngine    Historical replay with fill simulation
│   ├── RiskGuard         Pre-trade order validation (10 rules)
│   └── TWAP/VWAP/Iceberg Execution algorithms
│
├── Typed Domain Model (NEW in v3.0.0)
│   ├── Price             Decimal-backed, fail-fast on NaN/Inf
│   ├── Quantity           Non-negative, precision-safe
│   ├── Money              Currency-tracked arithmetic
│   └── Events            BarEvent, FillEvent, PositionEvent, ...
│
├── risk               Portfolio risk analytics (numpy/scipy)
│   ├── compute()      Full risk snapshot
│   ├── monte_carlo()  Monte Carlo VaR (Rust-accelerated)
│   ├── sharpe_inference() Lopez de Prado framework
│   ├── stress_test_all()  7 historical crisis scenarios
│   └── [20+ more methods]
│
├── execution          Multi-broker trading
│   ├── use_broker()   Configure broker (Alpaca, IBKR, SnapTrade, Polymarket)
│   ├── order()        Place orders with risk guard validation
│   ├── set_risk_guard() Attach pre-trade risk rules
│   └── get_positions() Current positions
│
├── data               Market and reference data (18 providers)
│   ├── bars()         OHLCV price data
│   ├── history()      Multi-symbol DataFrames
│   ├── economic()     FRED (800K+ series)
│   └── [indicators]   100+ technical indicators via TwelveData
│
├── indicators         25+ local indicators (no API calls)
├── signals            Position sizing and signal filters
├── portfolio          12 portfolio optimizers
├── alpha              IC analysis, signal decay, quantile returns
│
├── simons             Quantitative trading strategist AI
└── engine             Rust HFT engine for autonomous execution

Configuration

Get credentials at ai.cpz-lab.com/settings.

Testing

pytest --cov=cpz --cov-report=term-missing

Support

• Platform: ai.cpz-lab.com
• Repository: github.com/CPZ-Lab/cpz-py
• Email: contact@cpz-lab.com

_{Built by CPZ}