Python Bindings¶

import axon_quant.rl as rl

# Create trading environment
env = rl.TradingEnv(
    config={"initial_capital": 100_000.0, "max_steps": 500},
    market_data=bars,
    action_space={"type": "continuous", "min": -1.0, "max": 1.0},
    reward="sharpe",
)

# Standard Gymnasium interface
obs = env.reset()
obs, reward, terminated, truncated, info = env.step([0.5])

import axon_quant.llm as llm

# Create LLM backend
backend = llm.make_backend({
    "backends": [
        {"type": "openai_compat", "api_key": "your-key", "model": "gpt-4"}
    ]
})

# Create ReAct agent
agent = llm.ReActAgent(backend=backend, tools=[...])

# Run trading loop
response = agent.chat([llm.LLMMessage.user("Analyze BTC market")])

import axon_quant.hpo as hpo

# Define search space
search_space = {
    "learning_rate": hpo.SearchSpaceDef(param_type="log_uniform", low=1e-5, high=1e-3),
    "gamma": hpo.SearchSpaceDef(param_type="uniform", low=0.95, high=0.999),
}

# Create HPO runner
runner = hpo.OptunaHPO(
    search_space=search_space,
    objective_fn=objective_fn,
    study_name="ppo_optimization",
    directions=["maximize", "maximize"],
)

# Run optimization
results = runner.run(n_trials=50)

import axon_quant.tracker as tracker

# Create MLflow tracker
t = tracker.MLflowTracker(
    tracking_uri="http://localhost:5000",
    experiment_name="ppo_btc",
)

# Log parameters and metrics
t.log_param("learning_rate", 3e-4)
t.log_metric("sharpe_ratio", 1.5, step=100)

# Finish run
t.finish(tracker.RunStatus.Success)

import axon_quant.registry as registry

# Create registry
storage = registry.LocalStorage.new(base_dir="./models")
reg = registry.ModelRegistry.new(storage)

# Register model
version = reg.register(
    name="ppo_btc",
    artifact_path="./model.onnx",
    metadata={"algorithm": "PPO", "sharpe": "1.5"},
)

# Promote to production
reg.transition_stage("ppo_btc", version.version, registry.ModelStage.PRODUCTION)

from axon_quant.inference import (
    InferenceEngine, ModelConfig, Device, Observation, InferenceBackend,
    create_onnx_engine,
)

# One-step: create + load
engine = create_onnx_engine(
    model_path="model.onnx",
    input_shape=(1, 64, 128),
    output_dim=3,
)

# Single inference
obs = Observation(symbol="BTC-USDT", timestamp_ns=1_000_000_000, features=[0.0] * 128)
action = engine.infer(obs)
print(action.action_type, action.confidence, action.target_position)

from axon_quant.inference import create_onnx_engine, Observation

engine = create_onnx_engine(model_path="model.onnx", input_shape=(1, 64, 128), output_dim=3)
obs_list = [Observation(symbol="BTC-USDT", timestamp_ns=i * 1_000, features=[0.0] * 128) for i in range(32)]
actions = engine.infer_batch(obs_list)
assert len(actions) == 32

from axon_quant.inference import (
    BatchInferencePipeline, BatchConfig, create_onnx_engine, Observation,
)

engine = create_onnx_engine(model_path="model.onnx", input_shape=(1, 64, 128), output_dim=3)
bcfg = BatchConfig(max_batch_size=32, collect_timeout_us=500, num_workers=2)
pipe = BatchInferencePipeline(bcfg, engine)

# Buffer observations, then trigger one batch inference
for i in range(32):
    pipe.submit(Observation(symbol="BTC-USDT", timestamp_ns=i * 1_000, features=[0.0] * 128))
print(pipe.pending())  # 32
actions = pipe.collect()
print(len(actions), pipe.stats().total_inferences)  # 32 32

from axon_quant.inference import InferenceEngine, ModelConfig, Device, InferenceBackend

# Default Onnx (Stage 6 default feature, no extra compile flags needed)
engine_onnx = InferenceEngine(ModelConfig(
    path="model.onnx", backend=InferenceBackend.Onnx, device=Device.cpu(),
    input_shape=(1, 64, 128), output_dim=3,
))

# Candle (pure Rust, no ONNX runtime needed) — requires compile-time
# `candle-backend` feature: `cargo build -p axon-inference --features
# python --features candle-backend`. If not compiled, `__new__` returns
# `InferenceError("Candle backend not compiled: ...")`.
try:
    engine_candle = InferenceEngine(ModelConfig(
        path="model.safetensors", backend=InferenceBackend.Candle, device=Device.cpu(),
        input_shape=(1, 64, 128), output_dim=3,
    ))
except Exception as e:  # candle-backend feature not enabled
    print(f"skip: {e}")

from axon_quant.inference import InferenceEngine, ModelConfig, InferenceError, InferenceBackend, Device

cfg = ModelConfig(
    path="/nonexistent.onnx", backend=InferenceBackend.Onnx, device=Device.cpu(),
    input_shape=(1, 64, 128), output_dim=3,
)

try:
    engine = InferenceEngine(cfg)
    engine.load("/nonexistent.onnx")
except InferenceError as e:
    # e.args[0] is the stable error code (e.g. "ModelNotFound")
    # e.args[1] is the human-readable form: "[ModelNotFound] model file not found: /nonexistent.onnx"
    print(e.args[0], e.args[1])

import tempfile
from axon_quant.compliance import ComplianceModule, ComplianceConfig

tmp = tempfile.mkdtemp()
cfg = ComplianceConfig(
    account_id="acc-001",
    base_currency="USDT",
    large_trade_threshold=100_000.0,
    position_limit=1_000_000.0,
    max_portfolio_concentration=0.4,
    data_retention_years=7,
    regulators=["SEC", "FINRA"],
)
cm = ComplianceModule(cfg, tmp)

# Record trade (dict protocol; enum strings are case-insensitive)
cm.record_trade({
    "strategy_id": "strat-1",
    "symbol": "BTCUSDT",
    "side": "buy",
    "quantity": 1.0,
    "price": 50_000.0,
    "fee": 50.0,
    "fee_currency": "USDT",
    "exchange": "Binance",
})

print(cm.trade_count, cm.audit_event_count)  # 1 1
print(cm.verify_audit_integrity())  # True

# Query trades (all filters optional)
btc_trades = cm.query_trades({
    "symbol": "BTCUSDT",
    "side": "buy",
    "min_notional": 10_000.0,
    "start_time": "2026-01-01T00:00:00Z",
    "end_time": "2026-12-31T23:59:59Z",
})

# Stats (dict return)
stats = cm.get_trade_stats("2026-01-01T00:00:00Z", "2026-12-31T23:59:59Z")
print(stats["total_trades"], stats["win_rate"], stats["avg_trade_size"])

# Daily report (date="YYYY-MM-DD", starting_balance)
daily = cm.generate_daily_report("2026-06-24", 100_000.0)
print(daily["account_id"], daily["net_pnl"])

# Monthly report (year, month)
monthly = cm.generate_monthly_report(2026, 6)

# Annual report (year, initial_balance)
annual = cm.generate_annual_report(2026, 100_000.0)

from axon_quant.compliance import ComplianceModule, ComplianceConfig, ComplianceError
import tempfile

cfg = ComplianceConfig(
    account_id="acc-001", base_currency="USDT",
    large_trade_threshold=100_000.0, position_limit=1_000_000.0,
    max_portfolio_concentration=0.4, data_retention_years=7, regulators=["SEC"],
)
cm = ComplianceModule(cfg, tempfile.mkdtemp())

try:
    cm.record_trade({
        "strategy_id": "x", "symbol": "BTCUSDT", "side": "buy",
        "quantity": -1.0,  # triggers InvalidTradeData
        "price": 50_000.0, "fee": 50.0, "fee_currency": "USDT", "exchange": "Binance",
    })
except ComplianceError as e:
    # e.args[0] is the stable error code (e.g. "InvalidTradeData")
    # e.args[1] is the human-readable form: "[InvalidTradeData] Invalid trade data: Quantity must be positive"
    print(e.args[0], e.args[1])

import axon_quant.exchange as exchange

# Create Binance adapter
adapter = exchange.BinanceAdapter(
    api_key="your-key",
    api_secret="your-secret",
    testnet=True,
)

# Connect and subscribe
await adapter.connect()
await adapter.subscribe(["BTCUSDT"])

# Place order
order_id = await adapter.send_order(
    symbol="BTCUSDT",
    side="BUY",
    quantity=0.001,
    order_type="MARKET",
)

from axon_quant.backtest import (
    L1MatchingEngine, ImpactedMatchingEngineBuilder,
    BacktestEngine, limit_order,
)

# 1) Basic matching
engine = L1MatchingEngine()
engine.submit(limit_order(1, "BTC-USDT", "Sell", 100.0, 1.0))
result = engine.submit(limit_order(2, "BTC-USDT", "Buy", 100.0, 1.0))
print(result["is_filled"], len(result["fills"]))  # True, 1

# 2) Impact-aware (builder chain)
ie = (ImpactedMatchingEngineBuilder()
      .model_type("linear")
      .coefficient(0.1)
      .depth_levels(5)
      .build())
ie.submit(limit_order(3, "BTC-USDT", "Buy", 100.0, 1.0))
print(ie.permanent_offset())  # Cumulative permanent impact offset

# 3) Event-driven backtest
bt = BacktestEngine(initial_cash=100_000.0)
bt.push_event({
    "type": "order_submitted",
    "timestamp_ns": 1_000,
    "order": limit_order(1, "BTC-USDT", "Sell", 100.0, 1.0),
})
bt.push_event({
    "type": "order_submitted",
    "timestamp_ns": 2_000,
    "order": limit_order(2, "BTC-USDT", "Buy", 100.0, 1.0),
})
result = bt.run()
print(result.events_processed, result.fills, result.final_nav)

{
    "is_filled": bool,              # Fully filled
    "is_partially_filled": bool,    # Partially filled
    "remaining_quantity": float,    # Unfilled quantity
    "fills": [                      # List of fills
        {
            "fill_id": int,
            "taker_order_id": int,
            "maker_order_id": int,
            "price": float,
            "quantity": float,
            "taker_side": "BUY" | "SELL",  # Uppercase
        },
        ...
    ],
}

try:
    axon_quant.backtest.L1MatchingEngine().submit(bad_order)
except axon_quant.backtest.BacktestError as e:  # Actually an Exception subclass
    code = e.args[0]    # e.g. "Matching"
    msg = e.args[1]     # e.g. "[Matching] invalid side: xxx"

from axon_quant.risk import (
    DefaultRiskEngine, RiskConfig, CircuitBreaker,
    RiskResult, RiskReason, RiskMetrics, RiskError,
    make_order, make_portfolio, make_portfolio_with_positions,
    make_risk_config, make_circuit_breaker,
)

# 1) Create risk engine
engine = DefaultRiskEngine(make_risk_config(
    max_order_value=10_000.0,     # max order value
    max_leverage=2.0,              # max leverage multiplier
    max_daily_loss=5_000.0,        # max daily loss (triggers circuit breaker)
    max_concentration=0.30,        # max concentration of single instrument
))

# 2) Construct order + portfolio
order = make_order(
    id=1, symbol="BTC-USDT", side="Buy",
    type="limit", price=100.0, quantity=1.0,
)
portfolio = make_portfolio(
    base_currency="USD",
    commission_rate=0.001,
    cash={"USD": 100_000.0},
)

# 3) Pre-trade check
result = engine.check_order(order, portfolio)
if result.is_allow:
    print("Order allowed")
elif result.is_reject:
    reason = result.reason
    print(f"Rejected: {reason.kind}")  # e.g. "OrderTooLarge"
else:
    print(f"Warning: {result.message}")

# Cumulative daily loss exceeds threshold → engine.check_order() rejects
engine.update_daily_pnl(2_000.0)    # cumulative profit
engine.update_daily_pnl(-7_500.0)   # cumulative loss exceeds 5_000 → tripped
r = engine.check_order(order, portfolio)
assert r.is_reject and r.reason.kind == "CircuitBreakerActive"

# Reset daily state (does not reset VaR history window)
engine.reset_daily()

reason = RiskReason.from_dict({
    "kind": "OrderTooLarge",
    "max": 10_000.0,
    "actual": 20_000.0,
})
assert reason.kind == "OrderTooLarge"
assert reason.get("max") == 10_000.0
assert reason.get("actual") == 20_000.0
d = reason.to_dict()  # {"kind": "OrderTooLarge", "max": 10000.0, "actual": 20000.0}

cb = make_circuit_breaker(daily_loss_limit=10_000.0, cooldown_seconds=3600)
cb.check_and_trigger(-5_000.0)   # not at threshold, inactive
assert cb.is_active is False
cb.check_and_trigger(-15_000.0)  # triggered
assert cb.is_active is True
cb.reset()                       # force reset
assert cb.is_active is False

try:
    engine.check_order(bad_order, portfolio)
except axon_quant.risk.RiskError as e:  # actually Exception subclass
    code = e.args[0]   # e.g. "OrderRejected" / "CircuitBreakerActive"
    msg = e.args[1]    # e.g. "[OrderRejected] Order too large: ..."

from axon_quant.oms import (
    OrderManager, Order, OrderStatus, Side, OrderType, Portfolio, Position,
    OmsError, limit_order, market_order, make_order_status,
)

# 1) Create OMS + initial funding
mgr = OrderManager()
mgr.deposit("USDT", 100_000)

# 2) Submit order → returns order_id (UUID 36 chars)
oid = mgr.submit(limit_order(
    "BTC-USDT", "Buy", quantity=1, price=50_000,
    idempotency_key="my-bot-001",
))
print(oid, mgr.active_count())   # 1

# 3) Advance state machine
mgr.update_status(oid, make_order_status("Acknowledged"))

# 4) Push fill event (partial fill) → portfolio auto-updated
mgr.add_fill(
    order_id=oid, fill_id="f1", symbol="BTC-USDT",
    price=50_000, quantity=0.6, fee=0,
)
s = mgr.get_order_status(oid)
assert s.kind == "PartiallyFilled"
assert s.filled_qty == "0.6"

# 5) Query portfolio
snap = mgr.snapshot_balance()
assert snap["cash"]["USDT"] == "70000.0"
pos = snap["positions"]["BTC-USDT"]
assert pos.quantity == "0.6"
assert pos.avg_price == "50000"

# 6) Push completing fill → terminal state
mgr.add_fill(
    order_id=oid, fill_id="f2", symbol="BTC-USDT",
    price=51_000, quantity=0.4, fee=0,
)
assert mgr.get_order_status(oid) is None  # Filled removed from active

# Idempotency key dedup (re-submit with same key raises DuplicateIdempotencyKey)
oid_a = mgr.submit(limit_order("BTC-USDT", "Buy", 0.1, 50_000, idempotency_key="batch-1"))
try:
    oid_a2 = mgr.submit(limit_order("BTC-USDT", "Buy", 0.1, 50_000, idempotency_key="batch-1"))
except OmsError as e:
    assert e.args[0] == "DuplicateIdempotencyKey"

# Batch submit (loops submit, throws first error on failure)
oids = mgr.batch_submit([
    limit_order("ETH-USDT", "Buy", 1, 3_000, idempotency_key="batch-2"),
    limit_order("SOL-USDT", "Buy", 10, 100, idempotency_key="batch-3"),
])
assert len(oids) == 2

status = make_order_status("PartiallyFilled", filled_qty=0.6, avg_price=50_000)
assert status.kind == "PartiallyFilled"
assert status.filled_qty == "0.6"
assert status.avg_price == "50000"
assert status.is_terminal() is False

# Terminal predicate
filled = make_order_status("Filled", filled_qty=1, avg_price=50_000)
assert filled.is_terminal() is True
cancelled = make_order_status("Cancelled", reason="user_cancelled")
assert cancelled.is_terminal() is True

# Lightweight portfolio not depending on OrderManager (testing / serialization scenarios)
p = Portfolio()
p.deposit("USDT", 100_000)
p.deposit("BTC", 1.5)
p.apply_fill(
    fill_id="f1", symbol="BTC-USDT",
    price=50_000, quantity=0.6, fee=0,
)
assert p.cash["USDT"] == "70000.0"
assert p.positions["BTC-USDT"].quantity == "0.6"
assert p.position_count() == 1

# Serialize
d = p.to_dict()  # {"cash": {...}, "positions": {...}, "position_count": 1}

from decimal import Decimal

o = limit_order("BTC-USDT", "Buy", Decimal("0.1"), Decimal("50000.5"))
# quantity / price are also Decimal strings on the Python side

try:
    mgr.cancel("not-a-uuid")
except axon_quant.oms.OmsError as e:  # actually Exception subclass
    code = e.args[0]    # e.g. "OrderNotFound"
    msg = e.args[1]     # e.g. "[OrderNotFound] order not found: xxx"

import os
from axon_quant.exchange import BinanceAdapter, binance_testnet_config

# API key 自动从环境变量读取(BINANCE_API_KEY / BINANCE_API_SECRET)
# 缺一即抛 ExchangeError("BINANCE_API_KEY / BINANCE_API_SECRET not set in environment")
os.environ["BINANCE_API_KEY"] = "..."
os.environ["BINANCE_API_SECRET"] = "..."

adapter = BinanceAdapter(binance_testnet_config())
adapter.connect()  # 同步包装:内部 block_on 异步 connect
adapter.subscribe(symbols=["BTCUSDT"], kind="ticker")

# 下单:接受 dict,返回 order_id (UUID 字符串)
oid = adapter.place_order({
    "symbol": "BTCUSDT",
    "side": "buy",
    "type": "market",
    "quantity": "0.001",
    "tif": "IOC",
})

# 撤单
adapter.cancel_order(oid)

# 查询
balances = adapter.get_balance()
positions = adapter.get_positions()

from axon_quant.exchange import OrderLifecycleManager, ExchangeError

mgr = OrderLifecycleManager()
try:
    mgr.update_status(
        "00000000-0000-0000-0000-000000000000",
        {"status": "filled", "filled_qty": "0.1", "avg_price": "50000"},
    )
except ExchangeError as e:
    code = e.args[0]   # e.g. "OrderNotFound"
    msg  = e.args[1]   # e.g. "[OrderNotFound] order not found: ..."

adapter = BinanceAdapter(binance_testnet_config())
print(repr(adapter))   # "BinanceAdapter(...)"  — no secret
print(repr(config))    # "ExchangeConfig(Binance, testnet=True, rest=...)"  — no secret

import axon_quant

try:
    result = some_operation()
except axon_quant.AxonError as e:
    print(f"Error: {e}")
    print(f"Error type: {type(e).__name__}")

import asyncio
import axon_quant.exchange as exchange

async def main():
    adapter = exchange.BinanceAdapter(...)
    await adapter.connect()

    # Async operations
    balances = await adapter.get_balances()
    order_id = await adapter.place_order(...)

asyncio.run(main())

# Agent Swarm is currently implemented only in Rust layer
# Python bindings will be provided in future versions

from axon_quant._native.defi import (
    Chain, EvmConfig, DefiOrder, SwapRoute, RiskCheckResult,
    UniswapV3Contracts, DefiError,
)

# 1. Get chain config
chain = Chain.Ethereum
print(f"Chain: {chain.name}, ID: {chain.chain_id}")

# 2. Get Uniswap V3 contract addresses
contracts = UniswapV3Contracts.for_chain(Chain.Ethereum)
print(f"Router: {contracts.router}")

# 3. Create EVM config
config = EvmConfig(
    chain_id=1,
    rpc_url="https://mainnet.infura.io/v3/xxx",
    private_key="0x...",
)

# 4. Create DeFi order
order = DefiOrder("0xtoken", "1000", 50000.0)
print(f"Order: {order}")