Skip to content

Cookbook

Common patterns and recipes for using qlcore.

Converting Exchange Fills

Convert raw exchange data to qlcore fills:

import qlcore as qc

# Binance-style payload
raw_binance = {
    "id": "12345",
    "symbol": "BTCUSDT",
    "side": "BUY",
    "qty": "0.1",
    "price": "45000",
    "commission": "4.5",
    "time": 1701849600000,
}

fill = qc.Fill.from_exchange_fill("binance", raw_binance, market="perp")

# Coinbase-style payload
raw_coinbase = {
    "trade_id": "67890",
    "instrument": "BTC-USD",
    "side": "buy",
    "size": "0.1",
    "price": "45000",
    "fee": "4.5",
    "timestamp": 1701849600000,
}

fill = qc.Fill.from_exchange_fill("coinbase", raw_coinbase, market="spot")

Multi-Trade Portfolio PnL

Calculate portfolio PnL after multiple trades:

import qlcore as qc
from decimal import Decimal

# Setup
portfolio = qc.crypto_portfolio(initial_balance=10_000)

# Simulate trades
trades = [
    # Buy 0.1 BTC at 45000
    qc.Fill.create(
        order_id="1", instrument_id="BTC-USDT-PERP",
        side=qc.OrderSide.BUY, quantity="0.1",
        price="45000", fee="4.5", timestamp_ms=1000
    ),
    # Buy 0.1 BTC at 44000 (averaging down)
    qc.Fill.create(
        order_id="2", instrument_id="BTC-USDT-PERP",
        side=qc.OrderSide.BUY, quantity="0.1",
        price="44000", fee="4.4", timestamp_ms=2000
    ),
    # Sell 0.15 BTC at 46000 (partial close)
    qc.Fill.create(
        order_id="3", instrument_id="BTC-USDT-PERP",
        side=qc.OrderSide.SELL, quantity="0.15",
        price="46000", fee="6.9", timestamp_ms=3000
    ),
]

# Apply all trades
for fill in trades:
    portfolio = portfolio.apply_fill(fill)

# Calculate PnL at current price
prices = {"BTC-USDT-PERP": Decimal("47000")}
pnl = qc.calculate_portfolio_pnl(portfolio, prices)

print(f"Realized PnL: ${pnl.realized_pnl}")
print(f"Unrealized PnL: ${pnl.unrealized_pnl}")
print(f"Total Fees: ${pnl.total_fees}")
print(f"Net PnL: ${pnl.net_pnl}")

Tracking Funding Payments

Track funding on perpetual positions:

import qlcore as qc
from decimal import Decimal

# Create perpetual position
position = qc.PerpetualPosition(
    instrument_id="BTC-USDT-PERP",
    side=qc.PositionSide.LONG,
    size=Decimal("1.0"),
    avg_entry_price=Decimal("45000"),
    accumulated_funding=Decimal("0"),
)

# Funding events (every 8 hours)
funding_rates = [
    Decimal("0.0001"),   # Long pays $4.50
    Decimal("-0.0002"),  # Short pays, long receives $9.00
    Decimal("0.00005"),  # Long pays $2.25
]

mark_price = Decimal("45000")

for rate in funding_rates:
    payment = qc.calculate_funding_payment(
        position_size=position.size,
        mark_price=mark_price,
        funding_rate=rate,
    )
    position = position.apply_funding(payment)
    print(f"Funding Rate: {rate:.4%}, Payment: ${payment}, Total: ${position.accumulated_funding}")

# Check total realized including funding
print(f"Total Realized PnL: ${position.total_realized_pnl}")

Risk-Per-Trade Sizing

Size positions to risk a fixed percentage per trade:

import qlcore as qc
from decimal import Decimal

def calculate_position(
    equity: Decimal,
    risk_percent: Decimal,
    entry_price: Decimal,
    stop_price: Decimal,
    lot_size: Decimal = Decimal("0.001"),
):
    """Calculate position size with proper lot rounding."""

    # Calculate raw size
    raw_size = qc.risk_per_trade(
        equity=equity,
        risk_percent=str(risk_percent),
        entry_price=str(entry_price),
        stop_price=str(stop_price),
    )

    # Apply constraints
    final_size = qc.apply_position_limits(
        desired_size=raw_size,
        max_position_size=equity / entry_price,  # Max 100% equity
        min_position_size=lot_size,
        lot_size=lot_size,
    )

    # Risk validation
    max_loss = final_size * abs(entry_price - stop_price)
    risk_actual = max_loss / equity

    return {
        "size": final_size,
        "notional": final_size * entry_price,
        "max_loss": max_loss,
        "risk_percent": risk_actual,
    }

# Example
result = calculate_position(
    equity=Decimal("10000"),
    risk_percent=Decimal("0.02"),  # 2%
    entry_price=Decimal("45000"),
    stop_price=Decimal("44000"),
)

print(f"Position Size: {result['size']} BTC")
print(f"Notional Value: ${result['notional']}")
print(f"Max Loss: ${result['max_loss']}")
print(f"Risk %: {result['risk_percent']:.2%}")

Backtest vs Live Configuration

Use different configurations for backtesting and live:

import qlcore as qc

# Backtesting: optimize for speed
if mode == "backtest":
    config = qc.backtest_config()
    # {
    #     'cost_basis_method': 'AVERAGE',
    #     'track_lots': False,
    #     'validate_fills': True,
    #     'audit_logging': False,
    # }
    qc.disable_logging()  # Faster

# Live trading: optimize for safety
else:
    config = qc.live_config()
    # {
    #     'cost_basis_method': 'FIFO',
    #     'track_lots': True,
    #     'validate_fills': True,
    #     'audit_logging': True,
    #     'health_checks': True,
    # }
    qc.enable_logging()
    qc.set_log_level("INFO")

Event-Driven Architecture

Use event hooks for reactive patterns:

import qlcore as qc
from decimal import Decimal

# Global state
portfolio = qc.crypto_portfolio(initial_balance=10_000)

@qc.on("fill")
def handle_fill(fill):
    global portfolio
    portfolio = portfolio.apply_fill(fill)
    print(f"Fill applied: {fill.instrument_id} {fill.side} {fill.quantity}")

@qc.on("position_update")
def check_risk(position):
    if position.size > Decimal("1.0"):
        print(f"⚠️ Large position: {position.instrument_id}")

# In your trading loop
def on_trade_executed(raw_fill):
    fill = qc.Fill.from_dict(raw_fill)
    qc.emit("fill", fill)

    position = portfolio.positions.get(fill.instrument_id)
    if position:
        qc.emit("position_update", position)

Portfolio Persistence

Save and restore portfolio state:

import qlcore as qc
from pathlib import Path

PORTFOLIO_FILE = Path("portfolio_state.json")

def save_state(portfolio):
    """Save portfolio to disk."""
    qc.save_to_file(portfolio, str(PORTFOLIO_FILE))
    print(f"Portfolio saved to {PORTFOLIO_FILE}")

def load_state():
    """Load portfolio from disk."""
    if PORTFOLIO_FILE.exists():
        return qc.load_from_file(str(PORTFOLIO_FILE), qc.Portfolio)
    return qc.crypto_portfolio(initial_balance=10_000)

# Usage
portfolio = load_state()
# ... trading operations ...
save_state(portfolio)

Multi-Asset Risk Report

Generate a risk report across positions:

import qlcore as qc
from decimal import Decimal

def generate_risk_report(portfolio, prices):
    """Generate comprehensive risk report."""

    # Calculate portfolio metrics
    pnl = qc.calculate_portfolio_pnl(portfolio, prices)
    weights = qc.weights(portfolio, prices)

    # Exposure
    notionals = {
        k: p.size * prices[k] * (1 if p.side == qc.PositionSide.LONG else -1)
        for k, p in portfolio.positions.items()
    }
    net_exp = qc.net_exposure(notionals)
    gross_exp = qc.gross_exposure(notionals)

    # Report
    report = {
        "equity": sum(portfolio.account.balances.values()),
        "positions": len(portfolio.positions),
        "realized_pnl": pnl.realized_pnl,
        "unrealized_pnl": pnl.unrealized_pnl,
        "net_pnl": pnl.net_pnl,
        "total_fees": pnl.total_fees,
        "net_exposure": net_exp,
        "gross_exposure": gross_exp,
        "weights": weights,
    }

    return report

# Usage
prices = {
    "BTC-USDT-PERP": Decimal("46000"),
    "ETH-USDT-PERP": Decimal("2600"),
}

report = generate_risk_report(portfolio, prices)

print("=" * 50)
print("Risk Report")
print("=" * 50)
for key, value in report.items():
    print(f"{key}: {value}")