flox-py

Python bindings for FLOX — an AI-native framework for building trading systems.

AI agents discover the surface and drive it end-to-end through an MCP control plane. One strategy class runs backtest, paper, and live.

Install

pip install flox-py

Scaffold a new project

flox new my-strategy                                   # research scaffold (default)
flox new my-bot --template=live                        # live trading via CcxtBroker
flox new my-indicators --template=indicator-library    # publishable indicator package
flox templates                                         # list templates

flox new ships with the wheel; see docs/how-to/flox-new.md for what each template lays down.

Quick start

import flox_py as flox

registry = flox.SymbolRegistry()
btc = registry.add_symbol("binance", "BTCUSDT", tick_size=0.01)

class SMACross(flox.Strategy):
    def __init__(self, symbols):
        super().__init__(symbols)
        self.fast = flox.SMA(10)
        self.slow = flox.SMA(30)

    def on_trade(self, ctx, trade):
        f = self.fast.update(trade.price)
        s = self.slow.update(trade.price)
        if f is None or s is None:
            return
        if f > s and ctx.is_flat():
            self.market_buy(0.01)
        elif f < s and ctx.is_long():
            self.close_position()

def on_signal(sig):
    print(sig.side, sig.order_type, sig.quantity, sig.price)

runner = flox.Runner(registry, on_signal)
runner.add_strategy(SMACross([btc]))
runner.start()
# feed market data:
# runner.on_trade(btc, price, qty, is_buy, ts_ns)
# runner.on_book_snapshot(btc, bid_prices, bid_qtys, ask_prices, ask_qtys, ts_ns)
runner.stop()

Run a paper engine in one command

For tier-5/6 control (live order placement, position queries, kill switch over HTTP), flox engine sim boots a Runner + SimulatedExecutor + ControlServer + state-snapshot writer:

flox engine sim --strategy strategy.py --tape ./tape

Prints the engine URL and a copy-pasteable flox-mcp init --engine-url URL --token T command for wiring into AI tools.

AI companion

flox-mcp is a Model Context Protocol server that gives coding agents (Claude Code, Cursor, Cline) grounded access to indicators, error codes, the C-API surface, and full-text doc search:

pip install flox-mcp
flox-mcp init           # writes ./.mcp.json for the current project

See the flox-mcp README for the tool list.

Symbol and SymbolRegistry

registry = flox.SymbolRegistry()
btc = registry.add_symbol("binance", "BTCUSDT", tick_size=0.01)

btc.id        # int, e.g. 1
btc.name      # "BTCUSDT"
btc.exchange  # "binance"
btc.tick_size # 0.01

int(btc)   # 1
print(btc) # Symbol(binance:BTCUSDT, id=1)
# Symbol objects work transparently as int anywhere an ID is expected

Strategy

Subclass flox.Strategy and override the callbacks you need.

class MyStrategy(flox.Strategy):
    def __init__(self, symbols):
        super().__init__(symbols)   # symbols: list[Symbol | int]

    def on_start(self): ...
    def on_stop(self): ...

    def on_trade(self, ctx, trade): ...      # ctx: SymbolContext, trade: TradeData
    def on_book_update(self, ctx): ...
    def on_bar(self, ctx, bar): ...

    # Order-event hooks (fires on the strategy's own emitted orders).
    def on_fill(self, ctx, ev): ...           # status PARTIALLY_FILLED or FILLED
    def on_order_update(self, ctx, ev): ...   # every status change including fills

Order emission (shorthand, uses first symbol by default)

SymbolContext

TradeData

OrderEvent (passed to on_fill / on_order_update)

Runner

runner = flox.Runner(registry, on_signal)                  # synchronous
runner = flox.Runner(registry, on_signal, threaded=True)   # Disruptor background thread

runner.add_strategy(strategy)
runner.start()
runner.on_trade(btc, price, qty, is_buy, ts_ns)
runner.on_book_snapshot(btc, bid_prices, bid_qtys, ask_prices, ask_qtys, ts_ns)
runner.stop()

The on_signal callback receives Signal objects:

BacktestRunner

bt = flox.BacktestRunner(registry, fee_rate=0.0004, initial_capital=10_000)
bt.set_strategy(MyStrategy([btc]))

stats = bt.run_csv("data.csv")             # auto-detects symbol from registry
stats = bt.run_csv("data.csv", "BTCUSDT")  # explicit symbol name
stats = bt.run_tape("./tape")              # replay a recorded `.floxlog` directory

equity = bt.equity_curve()   # numpy arrays: timestamp_ns, equity, drawdown_pct
trades = bt.trades()         # numpy arrays: per-trade detail

from flox_py.report import write_html
write_html("report.html", stats=stats, equity_curve=equity, trades=trades)

The same risk-gate stack as the live Runner plugs into the backtest. Reduce-only / flatten orders bypass the gate by design (so a tightening cap cannot strand a position):

bt.set_risk_manager(my_risk_manager)        # IRiskManager: .allow(order)
bt.set_kill_switch(my_kill_switch)          # IKillSwitch: .check(order), .is_triggered()
bt.set_order_validator(my_validator)        # IOrderValidator: .validate(order, reason)
bt.set_pnl_tracker(my_pnl_tracker)          # IPnLTracker: .on_order_filled(order)

Stats dict

Walk-forward and grid search

wf = flox.WalkForwardRunner(
    registry, fee_rate=0.0004, initial_capital=10_000,
    mode="anchored", train_size=180, test_size=30,
)
wf.set_strategy_factory(lambda fold_idx: MyStrategy([btc]))
folds = wf.run_csv("data.csv", "BTCUSDT")

grid = flox.GridSearch()
grid.add_axis([5, 10, 20])    # fast period
grid.add_axis([30, 50, 100])  # slow period
def factory(params):
    fast, slow = params
    bt = flox.BacktestRunner(registry, fee_rate=0.0004, initial_capital=10_000)
    bt.set_strategy(MyStrategy([btc]))  # configure with fast / slow as needed
    return bt.run_csv("data.csv", "BTCUSDT")
grid.set_factory(factory)
results = grid.run()    # list of {index, params, stats}

Render a heatmap with flox_py.report.heatmap_html(...) / write_heatmap(...). Run a multiple-comparison-aware significance test with flox.whites_reality_check(returns, num_bootstrap=10_000).

MLflow

from flox_py import mlflow as flox_mlflow

flox_mlflow.log_backtest(
    stats, equity_curve=equity, trades=trades,
    params={"fast": 10, "slow": 30},
    run_name="sma-2025-01",
)

mlflow is optional — install with pip install mlflow.

Modules

All compute-heavy operations release the GIL for true parallelism.

Full API reference at flox-foundation.github.io/flox/reference/python.