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}")