API Reference¶

from axon_quant import (
    TradingEnv, EnvConfig,
    DefaultObservationSpace, FeatureConfig, FeatureSource, NormalizerType,
    DiscreteActionSpace, TradingDirection,
    PnLReward, SharpeReward, MultiObjectiveReward,
    MarketBar,
)

# 1. Configure environment
config = EnvConfig(
    initial_capital=100_000.0,    # Initial capital 100k USDT
    transaction_cost=0.001,       # Transaction cost 10 bps
    slippage=0.0005,              # Slippage 5 bps
    max_position_ratio=1.0,       # Maximum full position
    max_steps=1000,               # Maximum steps per episode
    seed=None,                    # Random seed
    symbol="BTCUSDT",             # Trading symbol
    return_window=252,            # Return history window (for Sharpe calculation)
)

# 2. Define observation space (feature engineering)
obs_space = DefaultObservationSpace.new(
    window_size=20,               # Keep last 20 time steps
    features=[
        FeatureConfig(
            name="close",
            source=FeatureSource.PriceField("close"),
            normalizer=NormalizerType.ZScore,  # Z-Score normalization
            clip_range=(-5.0, 5.0),            # Clip outliers
        ),
        FeatureConfig(
            name="volume",
            source=FeatureSource.VolumeField("volume"),
            normalizer=NormalizerType.ZScore,
        ),
        FeatureConfig(
            name="rsi",
            source=FeatureSource.RSI(14),      # Built-in RSI calculation
            normalizer=NormalizerType.MinMax,  # Map to [0, 1]
        ),
    ],
)

# 3. Define action space (discrete)
action_space = DiscreteActionSpace.new(
    n_quantity_bins=5,            # 5 position levels: 20%/40%/60%/80%/100%
    direction=TradingDirection.Both,  # Allow both long and short
)

# 4. Define reward function (multi-objective)
reward_fn = MultiObjectiveReward([
    PnLReward(relative=True, scale=1.0),     # Relative return
    SharpeReward(risk_free_rate=0.02, window=20),  # Rolling Sharpe ratio
])

# 5. Load market data
market_data = load_bars("BTCUSDT", "1h", start="2024-01-01", end="2024-06-01")

# 6. Create environment
env = TradingEnv.new(
    config=config,
    action_space=action_space,
    observation_space=obs_space,
    reward_fn=reward_fn,
    market_data=market_data,
)

# 7. Standard Gymnasium interaction loop
obs = env.reset()
done = False
total_reward = 0.0

while not done:
    # Can integrate RL model or rule-based strategy here
    action = model.predict(obs) if model else env.action_space.sample()

    obs, reward, done, info = env.step(action)
    total_reward += reward

    print(env.render())  # Output: step=123/5000 | value=$102340.50 | pos=0.5000

print(f"Episode total reward: {total_reward:.2f}")
print(f"Final portfolio value: {env.portfolio().portfolio_value:.2f}")

from axon_quant import LLMBackend, Message, ToolDefinition, LLMResponse

# Create OpenAI backend
llm = OpenAIBackend(
    api_key="YOUR_API_KEY",
    model="deepseek-chat",        # or "gpt-4", "claude-3-opus"
    base_url="https://api.deepseek.com",
)

# Basic conversation
messages = [
    Message(role="system", content="You are a professional quantitative trading analyst."),
    Message(role="user", content="Analyze BTC's current technical indicators."),
]

response = await llm.complete(messages)
print(response.content)

# Function Calling (tool use)
tools = [
    ToolDefinition(
        name="get_price",
        description="Get current price for specified trading pair",
        parameters={
            "type": "object",
            "properties": {
                "symbol": {"type": "string", "description": "Trading pair, e.g. BTCUSDT"},
            },
            "required": ["symbol"],
        },
    ),
    ToolDefinition(
        name="get_rsi",
        description="Calculate RSI indicator for specified trading pair",
        parameters={
            "type": "object",
            "properties": {
                "symbol": {"type": "string"},
                "period": {"type": "integer", "default": 14},
            },
            "required": ["symbol"],
        },
    ),
]

response = await llm.complete_with_tools(messages, tools)

# Parse tool calls
if response.tool_calls:
    for call in response.tool_calls:
        print(f"Calling tool: {call.name}, arguments: {call.arguments}")
        # Execute tool and return result...

# Error handling
from axon_quant import LLMError

try:
    response = await llm.complete(messages)
except LLMError.RateLimited as e:
    print(f"Rate limited, recommended wait {e.retry_after} seconds")
    await asyncio.sleep(e.retry_after or 60)
except LLMError.ContextOverflow as e:
    print(f"Context overflow: {e.needed} > {e.limit}")
    # Truncate history messages or switch to long-context model

from axon_quant import ExperimentTracker, MLflowTracker, MemoryTracker, ParamValue, RunStatus

# Create MLflow tracker (production)
tracker = MLflowTracker(
    tracking_uri="http://localhost:5000",
    experiment_name="ppo_btc_trading",
    run_name="run_2024_06_18_v1",
)

# Or create memory tracker (testing/fast iteration)
# tracker = MemoryTracker.new()

# Log hyperparameters
tracker.log_param("learning_rate", ParamValue.Float(3e-4))
tracker.log_param("batch_size", ParamValue.Int(128))
tracker.log_param("hidden_size", ParamValue.Int(256))
tracker.log_param("env_symbol", ParamValue.String("BTCUSDT"))

# Batch log parameters
tracker.log_params([
    ("gamma", ParamValue.Float(0.99)),
    ("gae_lambda", ParamValue.Float(0.95)),
    ("clip_range", ParamValue.Float(0.2)),
])

# Log metrics (supports step-based logging)
for step in range(1000):
    loss = train_step()
    tracker.log_metric("loss", loss, step=step)

    if step % 100 == 0:
        sharpe = evaluate_sharpe()
        tracker.log_metric("sharpe_ratio", sharpe, step=step)
        tracker.log_metric("portfolio_value", env.portfolio().portfolio_value, step=step)

# Log histogram (e.g., weight distribution)
tracker.log_histogram("actor_weights", weights_flattened, step=1000)

# Log image (e.g., PnL curve)
tracker.log_image("pnl_curve", png_bytes, format=ImageFormat.PNG, step=1000)

# Upload model artifact
tracker.log_artifact("model.onnx", Path("./models/model.onnx"))

# Set tags
tracker.set_tag("model_type", "PPO")
tracker.set_tag("data_source", "binance_1h")

# Finish run
tracker.finish(RunStatus.Success)

# Flush buffer (ensure data is written)
tracker.flush()

from axon_quant import (
    ModelRegistry, LocalStorage,
    ModelMetadata, ModelStage, SemVer, ModelSignature,
    VersionFilter,
)
from pathlib import Path

# Create registry (local file storage)
storage = LocalStorage.new(base_dir="./model_registry")
registry = ModelRegistry.new(storage)

# Register new model version
metadata = ModelMetadata(
    tags={
        "algorithm": "PPO",
        "env": "BTCUSDT_1h",
        "sharpe": "1.85",
    },
    description="PPO model v3, optimized Sharpe ratio",
)

signature = ModelSignature(
    inputs=["observation: float32[1,64,128]"],
    outputs=["action_probs: float32[1,3]"],
)

model_version = await registry.register(
    name="ppo_btc_trading",
    artifact_path=Path("./models/ppo_v3.onnx"),
    metadata=metadata,
    signature=signature,
)
print(f"Registration successful: {model_version.name}@{model_version.version}")
# Output: ppo_btc_trading@1.0.0

# Get latest version
latest = await registry.get("ppo_btc_trading", version=None)
print(f"Latest version: {latest.version}, stage: {latest.stage}")

# Get Production version
prod = await registry.get_production("ppo_btc_trading")

# Stage transition: Staging -> Production
await registry.transition_stage(
    name="ppo_btc_trading",
    version=SemVer.parse("1.0.0"),
    new_stage=ModelStage.Production,
)
# Note: When promoting to Production, old Production version automatically demotes to Archived

# Query version list
versions = await registry.list_versions(
    name="ppo_btc_trading",
    filter=VersionFilter(
        stage=ModelStage.Production,
        min_version=SemVer.parse("1.0.0"),
        limit=10,
    ),
)

# Rollback to previous Production version
rolled_back = await registry.rollback("ppo_btc_trading")
print(f"Rolled back to: {rolled_back.version}")

# Download model artifact
await registry.download_artifact(
    name="ppo_btc_trading",
    version=SemVer.parse("1.0.0"),
    dest=Path("./downloads/ppo_v1.onnx"),
)

# List all models
models = registry.list_models()
print(f"Registered models: {models}")

from axon_quant import (
    InferenceEngine, OnnxBackend, CandleBackend, TchBackend,
    ModelConfig, Device, InferenceBackend,
    Observation, Action,
)
from pathlib import Path

# Common configuration
config = ModelConfig(
    path="models/trading_model.onnx",
    backend=InferenceBackend.ONNX,
    device=Device.CUDA(0),        # Use GPU 0
    input_shape=[1, 64, 128],     # [batch, seq_len, features]
    output_dim=3,                 # Buy / Sell / Hold
    fp16=True,                    # Enable FP16
    num_threads=4,                # CPU threads
)

# ONNX backend
engine = OnnxBackend(config)
engine.load(Path(config.path))

# Candle backend (pure Rust, no Python dependency)
candle_config = ModelConfig(
    path="models/trading_model.safetensors",
    backend=InferenceBackend.CANDLE,
    device=Device.CPU,
    input_shape=[1, 4, 1],        # input_dim = 1*4*1 = 4
    output_dim=3,
    fp16=False,
    num_threads=4,
)
candle_engine = CandleBackend(candle_config)
candle_engine.load(Path(candle_config.path))

# Single inference
obs = Observation(
    features=[0.5, -0.2, 1.1, 0.0, ...],  # 64*128=8192 dimensions
    feature_names=[...],
    timestamp=1234567890,
)
action = engine.infer(obs)
print(f"Predicted action: {action}")

# Batch inference (recommended for production)
observations = [obs1, obs2, obs3, obs4]
actions = engine.infer_batch(observations)
print(f"Batch prediction: {len(actions)} actions")

# Hot update (atomic session replacement)
from axon_quant import ModelHotReloader

reloader = ModelHotReloader(engine, config)
reloader.spawn_watcher()  # Start file watcher

# Manual trigger reload
new_version = await reloader.reload()
print(f"Model updated to version {new_version}")

# Subscribe to version changes
version_rx = reloader.subscribe()
await version_rx.changed()
print(f"New version detected: {version_rx.borrow()}")

from axon_quant import (
    BinanceAdapter, OkxAdapter,
    ExchangeConfig, ExchangeId,
    Symbol, Order, OrderId, OrderType, Side, TimeInForce,
    RateLimitConfig, ReconnectConfig,
    MarginType, PositionMode,
)
from decimal import Decimal

# Binance configuration
config = ExchangeConfig(
    exchange_id=ExchangeId.Binance,
    api_key="YOUR_API_KEY",
    api_secret="YOUR_API_SECRET",
    passphrase=None,
    testnet=True,
    rest_base_url="https://testnet.binance.vision",
    ws_url="wss://testnet.binance.vision/ws",
    rate_limit=RateLimitConfig(
        requests_per_second=10,
        orders_per_minute=60,
        ws_messages_per_second=50,
    ),
    reconnect=ReconnectConfig(
        max_retries=10,
        initial_backoff_ms=500,
        max_backoff_ms=30000,
        backoff_multiplier=2.0,
        circuit_breaker_threshold=5,
        circuit_breaker_reset_sec=60,
    ),
    position_endpoint="/fapi/v2/positionRisk",
    fapi_base_url="https://testnet.binancefuture.com",
)

# Create and connect
adapter = BinanceAdapter(config)
await adapter.connect()

# Subscribe to market data
await adapter.subscribe([Symbol("BTCUSDT"), Symbol("ETHUSDT")])

# Get market data channel
market_rx = adapter.market_data_rx()
while True:
    msg = await market_rx.recv()
    match msg.type:
        case "Ticker":
            print(f"[{msg.data.symbol}] Bid {msg.data.bid} / Ask {msg.data.ask}")
        case "Trade":
            print(f"Trade: {msg.data.price} x {msg.data.quantity}")

# Place order
order = Order(
    client_order_id=OrderId.new(),
    symbol=Symbol("BTCUSDT"),
    side=Side.Buy,
    order_type=OrderType.Market,
    price=None,
    quantity=Decimal("0.001"),
    time_in_force=TimeInForce.Gtc,
    exchange=ExchangeId.Binance,
    meta={"strategy": "momentum_v1"},
)
order_id = await adapter.send_order(order)

# Cancel order
await adapter.cancel_order(order_id)

# Futures operations
await adapter.set_leverage("BTCUSDT", leverage=10)
await adapter.set_margin_type("BTCUSDT", MarginType.Isolated)
await adapter.set_position_mode(hedge_mode=True)

# Query account
account = await adapter.get_account_info()
print(f"Total balance: {account.total_balance}, Available: {account.available_balance}")

# Query funding rate
funding = await adapter.get_funding_rate("BTCUSDT")
print(f"Funding rate: {funding.rate}, Next settlement: {funding.next_funding_ms}")

from axon_quant import ActionSpace, DiscreteActionSpace, ContinuousActionSpace, TradingDirection

# Discrete action space
discrete = ActionSpace.Discrete(
    DiscreteActionSpace.new(n_quantity_bins=5, direction=TradingDirection.Both)
)
# Action indices: 0=Hold, 1-5=Buy(20%-100%), 6-10=Sell(20%-100%)

# Continuous action space
continuous = ActionSpace.Continuous(
    ContinuousActionSpace.new(min=-1.0, max=1.0)
)
# -1.0 = Full short, 0.0 = No position, 1.0 = Full long

from axon_quant import NormalizerType

NormalizerType.ZScore    # (x - mean) / std, preserves historical statistics
NormalizerType.MinMax    # (x - min) / (max - min) -> [0, 1]
NormalizerType.Robust    # (x - median) / IQR, outlier resistant
NormalizerType.None      # No normalization

from axon_quant import ModelStage

ModelStage.Staging      # Newly registered, awaiting validation
ModelStage.Production   # Running in production
ModelStage.Archived     # Old version archived
ModelStage.RolledBack   # Rolled back

from axon_quant import OrderType, TimeInForce

OrderType.Limit         # Limit order
OrderType.Market        # Market order
OrderType.StopLoss      # Stop loss order
OrderType.StopLimit     # Stop limit order

TimeInForce.Gtc         # Good Till Cancelled
TimeInForce.Ioc         # Immediate Or Cancel
TimeInForce.Fok         # Fill Or Kill